组成部分说明:
RabbitMQ包括五种队列模式,简单队列、工作队列、发布/订阅、路由、主题。
生产者将消息发送到队列,消费者从队列获取消息。
连接工具类:
/**
* 连接工厂创建的工具类
*/
public class RabbitMqUtils {
public static Channel getChannel() throws Exception{
//创建一个工厂
ConnectionFactory factory = new ConnectionFactory();
//工厂ip 连接RabbitMQ的队列
factory.setHost("192.168.75.128");
//用户名
factory.setUsername("guest");
//密码
factory.setPassword("guest");
//创建连接
Connection connection = factory.newConnection();
//获取通道
return connection.createChannel();
}
}
生产者发送消息:
public class Send {
private final static String QUEUE_NAME = "simple_queue";
public static void main(String[] argv) throws Exception {
// 1、获取到通道
Channel channel = RabbitMqUtils.getChannel();
// 2、声明(创建)队列
/**
* 参数明细
* 1、queue 队列名称
* 2、durable 是否持久化,如果持久化,mq重启后队列还在
* 3、exclusive 是否独占连接,队列只允许在该连接中访问,
* 如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时队列的创建
* 4、autoDelete 自动删除,队列不再使用时是否自动删除此队列,
* 如果将此参数和exclusive参数设置为true就可以实现临时队列(队列不用了就自动删除)
* 5、arguments 参数,可以设置一个队列的扩展参数,比如:可设置存活时间
*/
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 3、消息内容
String message = "Hello World!";
// 4、向指定的队列中发送消息
/**
* 参数明细:
* 1、exchange,交换机,如果不指定将使用mq的默认交换机(设置为"")
* 2、routingKey,路由key,交换机根据路由key来将消息转发到指定的队列,
* 如果使用默认交换机,routingKey设置为队列的名称
* 3、props,消息的属性
* 4、body,消息内容
*/
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
System.out.println(" 简单模式发送信息 '" + message + "'");
}
}
控制台结果:
web管理页面:
消费者接收消息:
public class Receive {
private final static String QUEUE_NAME = "simple_queue";
public static void main(String[] argv) throws Exception {
//获取通道
Channel channel = RabbitMqUtils.getChannel();
//接收到消息回调此函数
DefaultConsumer consumer = new DefaultConsumer(channel){
/**
* 当接收到消息后此方法将被调用
* @param consumerTag 消费者标签,用来标识消费者的,在监听队列时设置channel.basicConsume
* @param envelope 信封,通过envelope
* @param properties 消息属性
* @param body 消息内容
* @throws IOException
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
//交换机
String exchange = envelope.getExchange();
//消息id,mq在channel中用来标识消息的id,可用于确认消息已接收
long deliveryTag = envelope.getDeliveryTag();
// body 即消息体
String msg = new String(body, StandardCharsets.UTF_8);
System.out.println(" 简单模式收到信息 : "+msg+", 交换机:"+exchange+", 消息id: "+deliveryTag);
}
};
// 监听队列,第二个参数:是否自动进行消息确认。
/**
* 参数明细:
* 1、queue 队列名称
* 2、autoAck 自动回复,当消费者接收到消息后要告诉mq消息已接收,
* 如果将此参数设置为tru表示会自动回复mq,如果设置为false需要手动应答
* 3、callback,消费方法,当消费者接收到消息要执行的方法
*/
channel.basicConsume(QUEUE_NAME, true, consumer);
}
}
控制台结果:
web管理页面:
两个消费端共同消费同一个队列中的消息,但是一个消息只能被一个消费者获取。
生产者生产多个消息:
/**
* 生产者 生产大量消息
*/
public class Task01 {
private static final String QUEUE_NAME = "work_queue";
public static void main(String[] args) throws Exception {
Channel channel = RabbitMqUtils.getChannel();
channel.queueDeclare(QUEUE_NAME,false,false,false,null);
//从控制台当中接受信息
Scanner scanner = new Scanner(System.in);
while (scanner.hasNext()){
String message = scanner.next();
channel.basicPublish("",QUEUE_NAME,null, message.getBytes());
System.out.println("发送消息完成:"+message);
}
}
}
消费者启动两个线程:
/**
* 工作线程
*/
public class Worker01 {
//队列名称
public static final String QUEUE_NAME = "work_queue";
public static void main(String[] args) throws Exception {
Channel channel = RabbitMqUtils.getChannel();
//发送成功的消息回调
DeliverCallback deliverCallback = (consumerTag, message) -> {
System.out.println("接收到的消息:"+new String(message.getBody()));
};
//取消消息的回调
CancelCallback cancelCallback = consumerTag -> {
System.out.println(consumerTag+"消费者取消消费接口回调逻辑");
};
System.out.println("消费者1/2 等待接受消息...");
channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);
}
}
控制台:
根据结果发现,默认情况,mq是轮询分发的
通过 BasicQos 方法设置prefetchCount = 1。这样RabbitMQ就会使得每个Consumer在同一个时间点最多处理1个Message。换句话说,在接收到该Consumer的ack前,他它不会将新的Message分发给它。相反,它会将其分派给不是仍然忙碌的下一个Consumer。
prefetchCount在手动ack的情况下才生效,自动ack不生效。
x:交换机类型:Fanout,也称为广播
发送消息到交换机,交换机默认绑定到每个队列中,故所有消息都会发送到每个通道,实现一个消息被多个消费者消费。
生产者:
/**
* 声明Exchange,不再声明Queue
* 消费者中声明Queue
*/
public class Send {
private final static String EXCHANGE_NAME = "fanout_exchange";
public static void main(String[] argv) throws Exception {
// 获取通道
Channel channel = RabbitMqUtils.getChannel();
// 声明exchange,指定类型为fanout
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
// 消息内容
String message = "注册成功";
// 发布消息到Exchange
channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
System.out.println("生产者广播信息 '" + message + "'");
}
}
消费者1:
public class Receive {
private final static String QUEUE_NAME = "fanout_exchange_queue1";//消费者1队列
private final static String EXCHANGE_NAME = "fanout_exchange";
public static void main(String[] argv) throws Exception {
// 获取通道
Channel channel = RabbitMqUtils.getChannel();
// 声明队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 绑定队列到交换机
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
// 定义队列的消费者
DefaultConsumer consumer = new DefaultConsumer(channel) {
// 获取消息,并且处理,这个方法类似事件监听,如果有消息的时候,会被自动调用
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
byte[] body) throws IOException {
System.out.println(" 消费者1收到信息 : " + new String(body) + "!");
}
};
// 监听队列,自动返回完成
channel.basicConsume(QUEUE_NAME, true, consumer);
}
}
消费者2和1类似。
控制台:
生产者发送消息,消费者1、2都可以收到消息。
与路由模式不同的是,路由模式发送者只能精准匹配一个routing key,而主题模式可以实现发送者匹配多个队列
文章浏览阅读390次。SpringSpring Bean 的作用域有哪些?它的注册方式有几种?Spring 容器中管理一个或多个 Bean,这些 Bean 的定义表示为 BeanDefinition 对象,具体包含以下重要信息:Bean 的实际实现类;Bean 的引用或者依赖项;Bean 的作用范围;singleton:单例(默认);prototype:原型,每次调用bean都会创建新实例;request:每次http请求都会创建新的bean;session:同一个http session共享一个bean_spring和springboot的常见面试题
文章浏览阅读1.9k次。Openstack认证服务(认证组件)3_openstack 认证服务
文章浏览阅读4.5k次。第11节:人与人的相处(1) 2006-6-7 8∶40∶00 人与人的相处 一、有后台的下属。 我遇见过,也处理得很好。你不能得罪他背后的人,那么就通过他去利用他背后的人。比如说他是老板的亲戚,碰见别的部门有什么搞不定的人,你美言他几句叫他去搞,成功了自然是别人给老板面子,失败了你也可以多多积累他的错误,日后真到不得不踢人的时候也派得上..._外企重视documentation
文章浏览阅读3.4k次。这个应用程序是用iOS 15.5 SDK构建的。从2023年4月开始,所有提交到应用商店的iOS应用程序都必须使用iOS 16.1 SDK或更高版本构建,包括在Xcode 14.1或更高版本中。目前iOS 开发工具Xcode 版本号是13.4.1 ,系统无法升级,也会导致Xcode无法升级。1、苹果官方提示: 2023年4月开始,开发必须使用 Xcode 14.1 以上的版本,2、目前此电脑无法在升级, 2023年4月开始 ,此电脑就无法正常开发使用,应用程序商店连接操作错误。_sdk version issue. this app was built with the ios 15.5 sdk. all ios and ipa
文章浏览阅读335次。接单平台汇总程序员客栈码市开源众包智慧外包实现网猿急送人人开发网开发邦点鸭网快码网英选网外包大师我爱方案网智筹网自由智客接单注意事项:1、没有第三方担保的个人单子,尽量少接2、无需求文档、没有具体要求的不接3、没有预付的不做,尽量用442的分步步骤方式4、没有金刚钻,别揽瓷器活5、急单勿接6、任何不付定金的单子都是耍赖7、不计得失,不怕吃亏..._excel接单平台
文章浏览阅读1k次。烤CPU的时候,占用率满了,CPU频率的槽有一些还是空的…… 有没有能跑分的软件?好像有的【聊电Jing】你的CPU性能如何? 来跑个分测试看看吧! | Cinebench R15 & R20 使用教学_哔哩哔哩_bilibili 好像还是免费的Cinebench - Maxon Cinebench - Microsoft Store Apps 频率为什么就是超不过3Ghz? 多核,100度了? 可能频率最高只能这么高,再高可能就烧掉了…… 多核结果.................._cpu跑分教程
文章浏览阅读348次。Linux常用命令(1)切换到ROOT用户(su - root)[liu@localhost ~]$ su - root密码:[root@localhost ~]查看IP地址(ifconfig)[root@localhost ~]# ifconfigens33: flags=4163<UP,BROADCAST,RUNNING,MULTICAST> mtu 1500 inet 192.168.100.47 netmask 255.255.255.0 broad_code=exited, status=0/success
文章浏览阅读1.3k次。直接上代码:fanweiss(){//画圈varaaa=this.gaojingDatadebuggervarmap=newBMap.Map("ydmap");//创建Map实例varmPoint=newBMap.Point(this.gaojingData.longitude,this.gaojingData.latitude);//中心点map.setMapStyle({style:"midni..._bmap.circle
文章浏览阅读1.4k次。VisualVM原插件地址是oracle的打不开,已经移到github上了,具体如下:介绍:https://visualvm.github.io/plugins.html下载地址:https://visualvm.github.io/pluginscenters.html 选择对应JDK版本下载即可! 注意事项:在使用Visual VM进行heapdump分析的时候,发..._visualvm 插件中心地址
文章浏览阅读8.8k次,点赞15次,收藏80次。understand 常用操作文章目录understand 常用操作简单介绍软件下载常用基本操作新建工程并添加现有文件如何找到自己当前想要去编辑的文件?如何在当前文件中找到你要编辑的函数?如何跳转到定义?查看当前文件的函数列表如何查看函数都被谁调用了?查看函数的调用逻辑如何查找如何找到函数的被调用图除此之外可以分析出代码的各种结构文本的编辑格式设置双屏一边看代码,一遍看代码地图简单介绍understand对分析代码有非常强的能力,完全可以代替sourceinsight,并且可以在linux上mac上使_understand代码
文章浏览阅读4.1k次。如果是update,delete类误操作且已经commit,优先考虑使用flashback query进行恢复。select * from test1 as of timestamp to_timestamp('2018-01-13 16:59:29','YYYY-MM-DD hh24:mi:ss');如果是drop或truncate table,则不能使用闪回查询,需要使用备库进行整库..._数据库 oracle时间戳闪回
文章浏览阅读660次。在本地运行django1.python3.42.安装django,安装特定版本pip3 install django==2.03.测试安装python3import djangoprint(django.get_version())4.django使用创建一个新目录test_djangopython -m django --version_docker django print