外贸网站建设十大标准,知名商城网站建设价格低,上弦 网站建设,扬州做网站公司哪家好写在前面
本文看下rabbit mq的基础概念以及使用。
1#xff1a;简单介绍
为了不同进程间通信的解耦#xff0c;出现了消息队列#xff0c;为了规范消息队列的具体实现#xff0c;Java制定了jms规范#xff0c;这是一套基于接口的规范#xff0c;因此是绑定语言的…写在前面
本文看下rabbit mq的基础概念以及使用。
1简单介绍
为了不同进程间通信的解耦出现了消息队列为了规范消息队列的具体实现Java制定了jms规范这是一套基于接口的规范因此是绑定语言的即只能通过Java语言来实现和使用与jms类似还有基于net的nms这也是一套规范接口只不过是基于.net开发语言的。不管是jms还是nms它们都有一个通病就是无法实现跨语言这个时候amqp就应用而生了可以将其理解为一种应用层的协议构建在tcp之上因此就可以实现跨语言的消息通信参考下图
amqp协议通信模型如下
2基础环境准备
2.1服务安装
参考docker安装rabbitmq 。
2.2创建Virtual host和用户
创建virtual host 创建admin用户 设置admin权限 添加成功:
3正戏
本文主要看其提供的5种队列如下图
3.1简单队列
简单队列就是一个生产者一个消费者的队列方式如下图
生产消息
public class SimpleSend {private final static String QUEUE_NAME q_test_01;public static void main(String[] argv) throws Exception {// 获取到连接以及mq通道Connection connection ConnectionUtil.getConnection();// 从连接中创建通道Channel channel connection.createChannel();// 声明创建队列channel.queueDeclare(QUEUE_NAME, false, false, false, null);// 消息内容String message Hello World come here!!!;channel.basicPublish(, QUEUE_NAME, null, message.getBytes());System.out.println( [x] Sent message );//关闭通道和连接channel.close();connection.close();}
}生产后 点击q_test_01:
消费消息
public class SimpleRecv {private final static String QUEUE_NAME q_test_01;public static void main(String[] argv) throws Exception {// 获取到连接以及mq通道Connection connection ConnectionUtil.getConnection();// 从连接中创建通道Channel channel connection.createChannel();// 声明队列channel.queueDeclare(QUEUE_NAME, false, false, false, null);// 定义队列的消费者QueueingConsumer consumer new QueueingConsumer(channel);// 监听队列 true 这里使用自动确认及消息消费默认消费成功这种方式效率高但是容易丢失消息// 如果某些场景允许部分消息丢失但是要求执行效率则可以考虑将该值设置为true否则设置为false即手动确认// 最周通过执行channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);完成手动确认多一次网络通信channel.basicConsume(QUEUE_NAME, true, consumer);// 获取消息while (true) {QueueingConsumer.Delivery delivery consumer.nextDelivery();String message new String(delivery.getBody());System.out.println( [x] Received message );}}
}输出
[INFO] --- exec-maven-plugin:3.1.0:exec (default-cli) rabbitmq-study ---[x] Received Hello World come here!!!3.2work队列
一个生产者多个消费者如下图 work队列看起来和简单队列相比只是多起了几个消费者而已。 生产者
public class WorkSend {private final static String QUEUE_NAME test_queue_work;public static void main(String[] argv) throws Exception {// 获取到连接以及mq通道Connection connection ConnectionUtil.getConnection();Channel channel connection.createChannel();// 声明队列channel.queueDeclare(QUEUE_NAME, false, false, false, null);for (int i 0; i 100; i) {// 消息内容String message i;channel.basicPublish(, QUEUE_NAME, null, message.getBytes());System.out.println( [x] Sent message );Thread.sleep(i * 10);}channel.close();connection.close();}
}消费者
public class WorkRecv {private final static String QUEUE_NAME test_queue_work;public static void main(String[] argv) throws Exception {// 获取到连接以及mq通道Connection connection ConnectionUtil.getConnection();Channel channel connection.createChannel();// 声明队列channel.queueDeclare(QUEUE_NAME, false, false, false, null);// 同一时刻服务器只会发一条消息给消费者// 该值设置为1配合手动确认则可以实现一条一条消费确认一条后消费下一条且多个消费者时谁的消费能力强谁消费的消息多// 消费者之间消费消息不相互影响channel.basicQos(1);// 定义队列的消费者QueueingConsumer consumer new QueueingConsumer(channel);// 监听队列false表示手动返回完成状态true表示自动channel.basicConsume(QUEUE_NAME, false, consumer);
// channel.basicConsume(QUEUE_NAME, true, consumer);// 获取消息while (true) {QueueingConsumer.Delivery delivery consumer.nextDelivery();String message new String(delivery.getBody());System.out.println( [y] Received message );//休眠Thread.sleep(10);// 返回确认状态注释掉表示使用自动确认模式channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);}}
}3.3订阅模式
这种这种模式可以实现一个消息同时被多个消费者消费广播但是具体的实现需要依赖于交换器exchange生产者端将消息发送到交换器之后消费者端只需要将某个消息队列绑定到交换器交换器会将消息发送到绑定的所有队列消费者端就可以从队列中获取到对应的消息但需要注意一个队列的消息还是只可以获取一次如下图们
生产者端
public class SubscribeSend {private final static String EXCHANGE_NAME test_exchange_fanout111;public static void main(String[] argv) throws Exception {// 获取到连接以及mq通道Connection connection ConnectionUtil.getConnection();Channel channel connection.createChannel();// 声明exchangechannel.exchangeDeclare(EXCHANGE_NAME, fanout);// 消息内容String message Hello World!;channel.basicPublish(EXCHANGE_NAME, , null, message.getBytes());System.out.println( [x] Sent message );channel.close();connection.close();}
}消费者端2个
public class SubscribeRecv {private final static String QUEUE_NAME test_queue_work1;private final static String EXCHANGE_NAME test_exchange_fanout111;public static void main(String[] argv) throws Exception {// 获取到连接以及mq通道Connection connection ConnectionUtil.getConnection();Channel channel connection.createChannel();// 声明队列channel.queueDeclare(QUEUE_NAME, false, false, false, null);// 绑定队列到交换机channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, );// 同一时刻服务器只会发一条消息给消费者channel.basicQos(1);// 定义队列的消费者QueueingConsumer consumer new QueueingConsumer(channel);// 监听队列手动返回完成channel.basicConsume(QUEUE_NAME, false, consumer);// 获取消息while (true) {QueueingConsumer.Delivery delivery consumer.nextDelivery();String message new String(delivery.getBody());System.out.println( [Recv] Received message );Thread.sleep(10);channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);}}
}public class SubscribeRecv2 {private final static String QUEUE_NAME test_queue_work2;private final static String EXCHANGE_NAME test_exchange_fanout;public static void main(String[] argv) throws Exception {// 获取到连接以及mq通道Connection connection ConnectionUtil.getConnection();Channel channel connection.createChannel();// 声明队列channel.queueDeclare(QUEUE_NAME, false, false, false, null);// 绑定队列到交换机channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, );// 同一时刻服务器只会发一条消息给消费者channel.basicQos(1);// 定义队列的消费者QueueingConsumer consumer new QueueingConsumer(channel);// 监听队列手动返回完成channel.basicConsume(QUEUE_NAME, false, consumer);// 获取消息while (true) {QueueingConsumer.Delivery delivery consumer.nextDelivery();String message new String(delivery.getBody());System.out.println( [Recv2] Received message );Thread.sleep(10);channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);}}
}启动2个消费者 启动生产者
[INFO] --- exec-maven-plugin:3.1.0:exec (default-cli) rabbitmq-study ---[Recv2] Received Hello World![INFO] --- exec-maven-plugin:3.1.0:exec (default-cli) rabbitmq-study ---[Recv] Received Hello World!接着看下转换器和队列的绑定关系 注意以下的问题 1:因为交换器本身不具备数据存储的能力所以如果是某个交换器上没有绑定任何的队列则该消息将会丢失。
2:因为交换器本身不存储数据所以在具体的消息队列绑定到交换器即消费者启动前不要生产消息到交换器否则消息将会丢失。3.4路由模式
这种方式类似于订阅模式也需要转换器作为中间商但是并不会直接无脑的发送消息而是会根据消费者额外指定的路由key生产者在向转换器发送消息时会带着routeKey消费者在消费消息时会指定自己期望的routeKey只有二者匹配时才会从队列中消费对应的消息l另外注意的时交换机类型设置为direct,如下图
生产者
public class RoutingSend {private final static String EXCHANGE_NAME test_exchange_direct123;public static void main(String[] argv) throws Exception {// 获取到连接以及mq通道Connection connection ConnectionUtil.getConnection();Channel channel connection.createChannel();// 声明exchange 设置交换器为direct即路由模式channel.exchangeDeclare(EXCHANGE_NAME, direct);// 消息内容String message Hello World!;// 设置路由key为update即只有对应队列上设置了update路由key的消费者才会消费到消息channel.basicPublish(EXCHANGE_NAME, update, null, message.getBytes());System.out.println( [x] Sent message );channel.close();connection.close();}
}2个消费者
public class RoutingRecv {private final static String QUEUE_NAME test_queue_work_route;private final static String EXCHANGE_NAME test_exchange_direct123;public static void main(String[] argv) throws Exception {// 获取到连接以及mq通道Connection connection ConnectionUtil.getConnection();Channel channel connection.createChannel();// 声明队列channel.queueDeclare(QUEUE_NAME, false, false, false, null);// 绑定队列到交换机 设置路由key为deleteupdatechannel.queueBind(QUEUE_NAME, EXCHANGE_NAME, delete);channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, update);// 同一时刻服务器只会发一条消息给消费者channel.basicQos(1);// 定义队列的消费者QueueingConsumer consumer new QueueingConsumer(channel);// 监听队列手动返回完成channel.basicConsume(QUEUE_NAME, false, consumer);// 获取消息while (true) {QueueingConsumer.Delivery delivery consumer.nextDelivery();String message new String(delivery.getBody());System.out.println( [x] Received message );Thread.sleep(10);channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);}}
}public class RoutingRecv2 {private final static String QUEUE_NAME test_queue_work_route2;private final static String EXCHANGE_NAME test_exchange_direct;public static void main(String[] argv) throws Exception {// 获取到连接以及mq通道Connection connection ConnectionUtil.getConnection();Channel channel connection.createChannel();// 声明队列channel.queueDeclare(QUEUE_NAME, false, false, false, null);// 绑定队列到交换机channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, delete);channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, insert);// 同一时刻服务器只会发一条消息给消费者channel.basicQos(1);// 定义队列的消费者QueueingConsumer consumer new QueueingConsumer(channel);// 监听队列手动返回完成channel.basicConsume(QUEUE_NAME, false, consumer);// 获取消息while (true) {QueueingConsumer.Delivery delivery consumer.nextDelivery();String message new String(delivery.getBody());System.out.println( [x] Received message );Thread.sleep(10);channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);}}
}先启动消费者再启动生产者测试可。
写在后面
参考文章列表
RabbitMQ使用教程(超详细)
docker安装rabbitmq
RabbitMq的一些概念JMS、AMQP、MQ 。
