RabbitMQ——消息发布订阅

消息订阅发布

一个生产者，多个消费者
每个消费者都有自己的队列
生产者，没有直接把消息发送到队列，而是先发送到交换机
每个队列都要绑定到交换机
生产者发送的消息经过交换机，到达队列，就能实现一个消息被多个消费者消费

话不多说，直接撸代码：

老套路，首先还是生产者：

public class Send {private static final String EXCHANGE_NAME = "test_exchange";public static void main(String[] args) throws IOException, TimeoutException {Connection connection = RabbitConnection.getConnection(); //看前面文章写的工具类Channel channel = connection.createChannel();channel.exchangeDeclare(EXCHANGE_NAME, "fanout"); //声明exchange，这里为什么是fanout待会下面解释//send a msgString msg = "hello exchange";channel.basicPublish(EXCHANGE_NAME, "", null, msg.getBytes());System.out.println("send:"+msg);channel.close();connection.close();}
}

消费者1：

public class ReceiveOne {private static final String QUEUE_NAME = "receive1_queue";  //这里我们重新申请一个新的队列private static final String EXCHANGE_NAME = "test_exchange";public static void main(String[] args) throws IOException, TimeoutException {Connection connection = RabbitConnection.getConnection();Channel channel = connection.createChannel();channel.queueDeclare(QUEUE_NAME, false, false, false, null);channel.basicQos(1);channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ""); //这里需要绑定exchange,其他的和前面的work_queue是一样的DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [ConsumerOne is] Received '" + message + "'");try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}finally {channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);  //这里是手动应答}};boolean autoAck = false;channel.basicConsume(QUEUE_NAME, autoAck, deliverCallback, consumerTag -> { });}
}

消费者2：

public class ReceiveTwo {private static final String QUEUE_NAME = "receive2_queue";private static final String EXCHANGE_NAME = "test_exchange";public static void main(String[] args) throws IOException, TimeoutException {Connection connection = RabbitConnection.getConnection();Channel channel = connection.createChannel();channel.queueDeclare(QUEUE_NAME, false, false, false, null);channel.basicQos(1);channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [ConsumerTwo is] Received '" + message + "'");try {Thread.sleep(2000);} catch (InterruptedException e) {e.printStackTrace();}finally {channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);  //这里是手动应答}};boolean autoAck = false;channel.basicConsume(QUEUE_NAME, autoAck, deliverCallback, consumerTag -> { });    }
}

我们启动send和两个消费者，消费者同时消费，同一个消息，

再回到控制台，发现绑定了两个队列：

关于send中那个fanout参数，有必要讲一下，它表示不处理路由键，具体路由键是怎么回事，可参考下面这个博客：

路由键

至此消息订阅发布模式结束。

RabbitMQ——消息发布订阅相关推荐

python消息订阅_python rabbitmq消息发布订阅
发送端:import pika import sys connection = pika.BlockingConnection(pika.ConnectionParameters( host='loc ...
spring boot 整合谷歌guava的EventBus 实现单机版的消息发布订阅
spring boot 整合谷歌guava的EventBus 实现单机版的消息发布订阅大型分布式系统,直接用mq解耦,那么单机系统怎么办,可以考虑用EventBus 用EventBus的好处也是异 ...
三、RabbitMQ消息发布时的权衡
目录标题 RabbitMQ消息发布时的权衡失败通知消息发布时的权衡: 加入事务发送方确认备用交换器总结 RabbitMQ消息发布时的权衡失败通知 mandatory 消息发布时的权衡: 加 ...
RabbitMq之发布订阅模式
这里写了一个简单的springboot的demo来处理RabbitMq的发布订阅添加pom依赖 <dependency><groupId>com.rabbitmq</g ...
RabbitMQ 之发布订阅模式
publish/subscribe 发布订阅模式中,生产者不再直接与队列绑定,而是将数据发送至交换机Exchange 交换机Exchange用于将数据按某种规则送入与之绑定的队列,进而供消费者使用. ...
Vue组件学习之事件总线和消息发布订阅
简介主要介绍事件总线的定义和编写方法和Vue是如何实现消息的订阅与发布的. 事件总线事件总线是组件间通信的一种方式,适用于任意组件间的通信,比如毫不相干的两个组件.父子组件间.后代组件等等,都能通 ...
Redis - 消息发布订阅机制
Redis 发布订阅(pub/sub)是一种消息通信模式:发送者(pub)发送消息,订阅者(sub)接收消息.发布者生产消息放到队列里,多个监听队列的消费者都会收到同一份消息. Redis客户端可以订 ...
rabbitmq消息发布mandatory参数
channel.basicPublish("", "",true, null, message.getBytes(StandardCharsets.UTF_8) ...
基于MQTT的消息发布订阅python实现
简介: MQTT 全称为 Message Queuing Telemetry Transport(消息队列遥测传输)是一种基于发布/订阅范式的"轻量级"消息协议.该协议构建于TCP ...

RabbitMQ——消息发布订阅

消息订阅发布

RabbitMQ——消息发布订阅相关推荐

最新文章

热门文章