（二）丶RabbitMQ的六大核心

一丶什么是MQ

Message Queue(消息队列）简称MQ，是一种应用程序对应用程序的消息通信机制。在MQ中，消息以队列形式存储，以便于异步传输，在MQ中，发布者（生产者）将消息放入队列，而消费者从队列中读取并处理这些消息，这种设计允许生产者和消费者之间解耦，提高系统的响应速度和吞吐量，MQ常用于解耦系统之间的依赖关系，提高系统的稳定性和可扩展性，MQ还支持消峰，即以稳定的系统资源应对突发的流量冲剂，然而使用MQ也可能带来一些挑战，如：系统可用性降低、系统复杂度提高、以及消息一致性问题等；

二丶常见MQ有哪些

目前业界有很多的MQ产品，例如RabbitMQ、RocketMQ、Kafka、ZeroMQ、MetaMQ等，也有直接使用Redis充当消息队列的案列，而这些消息队列产品，各有侧重，在实际选型时，需要结合自身需求及MQ的产品特征等，综合考虑；

RabbitMQ是实现了高级消息队列协议(AMQP)的开源消息代理软件（亦称面向消息的中间件）。RabbitMQ服务器是Erlang语言编写，而集群和故障转移是构建在开放电信平台框架上的。所有的主要变成语言均有与代理接口通讯的客户端库。

三丶RabbitMQ六种消息模式

1.Simple Work Queue（简单工作队列）：常见的一对一模式，一条消息由一个消费者进行消费。如果有多个消费者，默认是使用轮询的方式将消息分配消费者

2.Work Queues（工作队列模式）：也叫公屏队列，能者多劳的消息队列模型。队列必须收到来自消费者的手动ACK（消息确认机制）才可以继续往消费者发送消息。

3.Publish/Subscribe（发布订阅模式）：一条消息被多和消费者消费。

4.Ruoting（路由模式）：有选择的接收消息。

5.Topics（主题模式）：通过一定的规则来选择性的接收消息

6.RPC模式：发布者发布消息，并且通过RPC方式等待结果。

（1）Simple Work Queue（简单工作队列）

消息生产后将消息放入队列，消息的消费者（consumer）监听消息队列嘛，如果队列中有消息就消费掉，消息被消费后自动从消息队列中删除。（也可能出现异常）

/*** @Description:获取RabbitMQ连接* @Author: xy丶*/
public class RabbitMQConnection {public final static String RABBITMQ_SERVER_HOST = "192.168.0.122";public final static int RABBITMQ_SERVER_PORT = 5672;public final static String VIRTUAL_HOST = "/XY_HOST";public static Connection getConnection() throws IOException, TimeoutException {//1、创建连接ConnectionFactory factory = new ConnectionFactory();//2、设置主机名factory.setHost(RABBITMQ_SERVER_HOST);//3、设置通讯端口，默认是5672，不专门设置也可以factory.setPort(RABBITMQ_SERVER_PORT);//4、设置账号和密码factory.setUsername("admin");factory.setPassword("admin");//4、设置Virtual Hostfactory.setVirtualHost(VIRTUAL_HOST);//5、创建连接return factory.newConnection();}}

消费者

/*** @Description:简单工作队列模式消费者* @Author: xy丶*/
@Slf4j
public class Consumer {private final static String SIMPLE_WORK_QUEUE= "simple_work_queue";public static void main(String[] args) throws IOException, InterruptedException, TimeoutException {//获取连接Connection connection = RabbitMQConnection.getConnection();//获取通道Channel channel = connection.createChannel();channel.queueDeclare(SIMPLE_WORK_QUEUE, false, false, false, null);QueueingConsumer consumer = new QueueingConsumer(channel);channel.basicConsume(SIMPLE_WORK_QUEUE, true, consumer);QueueingConsumer.Delivery delivery = consumer.nextDelivery();String message = new String(delivery.getBody());log.info("Consumer reception "+message);}
}

生产者

/*** @Description:简单工作队列模式生产者* @Author: xy丶*/
@Slf4j
public class Producer {private final static String SIMPLE_WORK_QUEUE = "simple_work_queue";/*** 简单工作队列模式* @param args* @throws IOException* @throws TimeoutException*/public static void main(String[] args) throws IOException, TimeoutException {//创建连接Connection connection = RabbitMQConnection.getConnection();//创建通道Channel channel = connection.createChannel();// 声明队列 String var1, 是否持久化// boolean var2, 是否排外 只允许该channel访问该队列 为true只能有一个消费者// boolean var3, 是否自动删除// boolean var4, 消费完删除// Map<String, Object> var5 其他属性channel.queueDeclare(SIMPLE_WORK_QUEUE, false, false, false, null);// 消息内容 String var1, 是否持久化// boolean var2, 是否排外 只允许该channel访问该队列 为true只能有一个消费者// boolean var3, 是否自动删除// boolean var4, 消费完删除// Map<String, Object> var5 其他属性String message = "Hello Word!!!";channel.basicPublish("", SIMPLE_WORK_QUEUE,null,message.getBytes());log.info("Producer send "+message);//最后关闭通关和连接channel.close();connection.close();}

(2).Work Queues（工作队列模式）创建生产者两个消费者看看效果

生产者

package com.puwang.MQ.workQueue;import com.puwang.MQ.config.RabbitMQConnection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import lombok.extern.slf4j.Slf4j;import java.io.IOException;
import java.util.Scanner;
import java.util.concurrent.TimeoutException;/*** @Description:工作队列模式生产者* @Author: xy丶*/
@Slf4j
public class Producer {private final static String QUEUE_WORK = "QUEUE_WORK";public static void main(String[] args) throws Exception {Connection connection = RabbitMQConnection.getConnection();Channel channel = connection.createChannel();channel.queueDeclare(QUEUE_WORK, false, false, false, null);for(int i = 0; i < 20; i++){String message = "娃哈哈" + i;channel.basicPublish("", QUEUE_WORK, null, message.getBytes());System.out.println("send=============="+message);Thread.sleep(i*10);}channel.close();connection.close();}
}

@Slf4j
public class WorkQueueConsumer1 {private final static  String QUEUE_WORK = "QUEUE_WORK";/*** 结果：** 1、一条消息只会被一个消费者接收；** 2、rabbit采用轮询的方式将消息是平均发送给消费者的；** 3、消费者在处理完某条消息后，才会收到下一条消息。* @param args* @throws IOException* @throws TimeoutException*/public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {Connection connection = RabbitMQConnection.getConnection();Channel channel = connection.createChannel();channel.queueDeclare(QUEUE_WORK, false,false, false,null);//同一时刻服务器只会发送一条消息给消费者channel.basicQos(1);QueueingConsumer consumer = new QueueingConsumer(channel);//关于手工确认 待之后有时间研究下channel.basicConsume(QUEUE_WORK, false, consumer);while(true){QueueingConsumer.Delivery delivery = consumer.nextDelivery();String message = new String(delivery.getBody());log.info("[消费者1] Received1 '"+message+"'");Thread.sleep(10);//返回确认状态channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);}}
}@Slf4j
public class WorkQueueConsumer2 {private final static  String QUEUE_WORK = "QUEUE_WORK";public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {Connection connection = RabbitMQConnection.getConnection();Channel channel = connection.createChannel();channel.queueDeclare(QUEUE_WORK, false,false, false,null);//同一时刻服务器只会发送一条消息给消费者channel.basicQos(1);QueueingConsumer consumer = new QueueingConsumer(channel);//关于手工确认 待之后有时间研究下channel.basicConsume(QUEUE_WORK, false, consumer);while(true){QueueingConsumer.Delivery delivery = consumer.nextDelivery();String message = new String(delivery.getBody());log.info("[消费者2] Received1 '"+message+"'");Thread.sleep(10);//返回确认状态channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);}}
}

(3).Publish/Subscribe（发布订阅模式）

生产者

/*** 订阅模式 生产者* 订阅模式：一个生产者发送的消息会被多个消费者获取。* 消息产生者将消息放入交换机,交换机发布订阅把消息发送到所有消息队列中,对应消息队列的消费者拿到消息进行消费* 相关场景:邮件群发,群聊天,广播(广告)* @Description:发布订阅模式生产者* @Author: xy丶*/
@Slf4j
public class Producer {private final static String PUBLISH_SUBSCRIBE_EXCHANGE = "Publish_subscribe_exchange";/*** 交换机发布订阅把消息发送到所有消息队列中,对应消息队列的消费者拿到消息进行消费* 相关场景:邮件群发,群聊天,广播(广告)* @param args*/public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {//获取连接Connection connection = RabbitMQConnection.getConnection();//从连接中获取一个通道Channel channel = connection.createChannel();消费者绑定交换机 参数1 队列 参数2 交换机 参数3 routingKeychannel.exchangeDeclare(PUBLISH_SUBSCRIBE_EXCHANGE, "fanout");//发送消息for (int i = 0; i < 10; i++) {String message = "哇哈哈哈！！！"+i;log.info("send message：" + message);//发送消息channel.basicPublish(PUBLISH_SUBSCRIBE_EXCHANGE, "", null, message.getBytes("utf-8"));Thread.sleep(100 * i);}channel.close();connection.close();}
}

消费者

/*** @Description:发布订阅模式消费者* @Author: xy丶*/
@Slf4j
public class PublishSubscribeConsumer1 {//交换机名称private final static String PUBLISH_SUBSCRIBE_EXCHANGE = "Publish_subscribe_exchange";//队列名称private static final String PUBLISH_SUBSCRIBE_QUEUE    = "publish_subscribe_queue";public static void main(String[] args) throws IOException, TimeoutException {//获取连接Connection connection = RabbitMQConnection.getConnection();//从连接中获取一个通道final Channel channel = connection.createChannel();//声明交换机（分发:发布/订阅模式）channel.exchangeDeclare(PUBLISH_SUBSCRIBE_EXCHANGE, "fanout");//声明队列channel.queueDeclare(PUBLISH_SUBSCRIBE_QUEUE, false, false, false, null);//将队列绑定到交换机channel.queueBind(PUBLISH_SUBSCRIBE_QUEUE, PUBLISH_SUBSCRIBE_EXCHANGE, "");//保证一次只分发一个int prefetchCount = 1;channel.basicQos(prefetchCount);//定义消费者DefaultConsumer consumer = new DefaultConsumer(channel) {//当消息到达时执行回调方法@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String message = new String(body, "utf-8");log.info("[PublishSubscribeConsumer1] Receive message：" + message);try {//消费者休息2s处理业务Thread.sleep(1000);}catch (InterruptedException e) {e.printStackTrace();}finally {System.out.println("[1] done");//手动应答channel.basicAck(envelope.getDeliveryTag(), false);}}};//设置手动应答boolean autoAck = false;//监听队列channel.basicConsume(PUBLISH_SUBSCRIBE_QUEUE, autoAck, consumer);}
}/*** @Description:发布订阅模式消费者* @Author: xy丶*/
@Slf4j
public class PublishSubscribeConsumer1 {//交换机名称private final static String PUBLISH_SUBSCRIBE_EXCHANGE = "Publish_subscribe_exchange";//队列名称private static final String PUBLISH_SUBSCRIBE_QUEUE    = "publish_subscribe_queue";public static void main(String[] args) throws IOException, TimeoutException {//获取连接Connection connection = RabbitMQConnection.getConnection();//从连接中获取一个通道final Channel channel = connection.createChannel();//声明交换机（分发:发布/订阅模式）channel.exchangeDeclare(PUBLISH_SUBSCRIBE_EXCHANGE, "fanout");//声明队列channel.queueDeclare(PUBLISH_SUBSCRIBE_QUEUE, false, false, false, null);//将队列绑定到交换机channel.queueBind(PUBLISH_SUBSCRIBE_QUEUE, PUBLISH_SUBSCRIBE_EXCHANGE, "");//保证一次只分发一个int prefetchCount = 1;channel.basicQos(prefetchCount);//定义消费者DefaultConsumer consumer = new DefaultConsumer(channel) {//当消息到达时执行回调方法@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String message = new String(body, "utf-8");log.info("[PublishSubscribeConsumer1] Receive message：" + message);try {//消费者休息2s处理业务Thread.sleep(1000);}catch (InterruptedException e) {e.printStackTrace();}finally {System.out.println("[1] done");//手动应答channel.basicAck(envelope.getDeliveryTag(), false);}}};//设置手动应答boolean autoAck = false;//监听队列channel.basicConsume(PUBLISH_SUBSCRIBE_QUEUE, autoAck, consumer);}
}/*** @Description:发布订阅模式消费者* @Author: xy丶*/
@Slf4j
public class PublishSubscribeConsumer2 {//交换机名称private final static String PUBLISH_SUBSCRIBE_EXCHANGE = "Publish_subscribe_exchange";//队列名称private static final String PUBLISH_SUBSCRIBE_QUEUE    = "publish_subscribe_queue";public static void main(String[] args) throws IOException, TimeoutException {//获取连接Connection connection = RabbitMQConnection.getConnection();//从连接中获取一个通道final Channel channel = connection.createChannel();//声明交换机（分发:发布/订阅模式）channel.exchangeDeclare(PUBLISH_SUBSCRIBE_EXCHANGE, "fanout");//声明队列channel.queueDeclare(PUBLISH_SUBSCRIBE_QUEUE, false, false, false, null);//将队列绑定到交换机channel.queueBind(PUBLISH_SUBSCRIBE_QUEUE, PUBLISH_SUBSCRIBE_EXCHANGE, "");//保证一次只分发一个int prefetchCount = 1;channel.basicQos(prefetchCount);//定义消费者DefaultConsumer consumer = new DefaultConsumer(channel) {//当消息到达时执行回调方法@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String message = new String(body, "utf-8");log.info("[PublishSubscribeConsumer2] Receive message：" + message);try {//消费者休息2s处理业务Thread.sleep(1000);}catch (InterruptedException e) {e.printStackTrace();}finally {System.out.println("[1] done");//手动应答channel.basicAck(envelope.getDeliveryTag(), false);}}};//设置手动应答boolean autoAck = false;//监听队列channel.basicConsume(PUBLISH_SUBSCRIBE_QUEUE, autoAck, consumer);}
}

(4).Ruoting（路由模式）

消费者

/*** 1）消息生产者将消息发送给交换机按照路由判断,路由是字符串(info) 当前产生的消息携带路由字符(对象的方法),交换机根据路由的key,只能匹配上路由key对应的消息队列,对应的消费者才能消费消息* 2）根据业务功能定义路由字符串* 3）从系统的代码逻辑中获取对应的功能字符串,将消息任务扔到对应的队列中业务场景:error 通知;EXCEPTION;错误通知的功能;传统意义的错误通知;* 客户通知;利用key路由,可以将程序中的错误封装成消息传入到消息队列中,开发者可以自定义消费者,实时接收错误* @Description:发布订阅模式生产者* @Author: xy丶*/
@Slf4j
public class Producer {//路由交换机名称static final String ROUTING_DIRECT_EXCHANGE = "Routing_direct_exchange";//队列名称1 发送static final String QUEUE_SEND = "queue_send";//队列名称2  接收static final String QUEUE_RECEIVE = "queue_receive";public static void main(String[] args) throws Exception {//创建连接Connection connection = RabbitMQConnection.getConnection();//创建频道Channel channel = connection.createChannel();/*** 声明交换机* 参数1：交换机名称* 参数2：交换机类型，fanout,toppic,direct,headers*/channel.exchangeDeclare(ROUTING_DIRECT_EXCHANGE, "direct");/*** 声明（创建）队列* 参数1：队列名称* 参数2：是否定义持久化队列* 参数3：是否独占本次连接* 参数4：是否在不使用的时候自动删除队列* 参数5：队列其它参数*/channel.queueDeclare(QUEUE_SEND,true,false,false,null);channel.queueDeclare(QUEUE_RECEIVE,true,false,false,null);/*** 队列绑定交换机* 参数1：队列名称* 参数2：是否定义持久化队列* 参数3：是否独占本次连接*/channel.queueBind(QUEUE_SEND,ROUTING_DIRECT_EXCHANGE,"send");channel.queueBind(QUEUE_RECEIVE,ROUTING_DIRECT_EXCHANGE,"receive");//发送消息String message = "路由模式：routing key 为 send";/*** 参数1：交换机名称，如果没有指定则使用默认Default Exchage* 参数2：路由key,简单模式可以传递队列名称* 参数3：消息其它属性* 参数4：消息内容*/channel.basicPublish(ROUTING_DIRECT_EXCHANGE,"send",null,message.getBytes());log.info("已发送消息："+message);//发送消息message = "路由模式：routing key 为 receive";/*** 参数1：交换机名称，如果没有指定则使用默认Default Exchage* 参数2：路由key,简单模式可以传递队列名称* 参数3：消息其它属性* 参数4：消息内容*/channel.basicPublish(ROUTING_DIRECT_EXCHANGE,"receive",null,message.getBytes());log.info("已发送消息："+message);//关闭资源channel.close();connection.close();}
}

消费者


/***路由消费者*/
@Slf4j
public class RoutingConsumer1 {//路由交换机名称static final String ROUTING_DIRECT_EXCHANGE = "Routing_direct_exchange";//队列名称1 发送static final String QUEUE_SEND = "queue_send";public static void main(String[] args) throws Exception {//创建连接Connection connection = RabbitMQConnection.getConnection();//创建通道（频道）Channel channel = connection.createChannel();//声明交换机channel.exchangeDeclare(ROUTING_DIRECT_EXCHANGE,"direct");/*** 声明（创建）队列* 参数1：队列名称* 参数2：是否定义持久化队列* 参数3：是否独占本次连接* 参数4：是否在不使用的时候自动删除队列* 参数5：队列其它参数*/channel.queueDeclare(QUEUE_SEND,true,false,false,null);//队列绑定交换机channel.queueBind(QUEUE_SEND, ROUTING_DIRECT_EXCHANGE,"send");//创建消费这，并设置消息处理DefaultConsumer consumer = new DefaultConsumer(channel) {/*** consumerTag 消息者标签，在channel.basicConsume时候可以指定* envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，*          消息和重传标志(收到消息失败后是否需要重新发送)* properties 属性信息* body 消息*/@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {//路由keylog.info("路由key为：" + envelope.getRoutingKey());//交换机log.info("交换机为：" + envelope.getExchange());//消息idlog.info("消息id为：" + envelope.getDeliveryTag());//收到的消息log.info("消费者1-接收到的消息为：" + new String(body, "utf8"));}};/*** 监听消息* 参数1：队列名称* 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认* 参数3：消息接收到后回调*/channel.basicConsume(QUEUE_SEND, true, consumer);}/***路由消费者*/
@Slf4j
public class RoutingConsumer2 {//路由交换机名称static final String ROUTING_DIRECT_EXCHANGE = "Routing_direct_exchange";//队列名称1 发送static final String QUEUE_RECEIVE = "queue_receive";public static void main(String[] args) throws Exception {//创建连接Connection connection = RabbitMQConnection.getConnection();//创建通道（频道）Channel channel = connection.createChannel();//声明交换机channel.exchangeDeclare(ROUTING_DIRECT_EXCHANGE,"direct");/*** 声明（创建）队列* 参数1：队列名称* 参数2：是否定义持久化队列* 参数3：是否独占本次连接* 参数4：是否在不使用的时候自动删除队列* 参数5：队列其它参数*/channel.queueDeclare(QUEUE_RECEIVE,true,false,false,null);//队列绑定交换机channel.queueBind(QUEUE_RECEIVE, ROUTING_DIRECT_EXCHANGE,"receive");//创建消费这，并设置消息处理DefaultConsumer consumer = new DefaultConsumer(channel) {/*** consumerTag 消息者标签，在channel.basicConsume时候可以指定* envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，*          消息和重传标志(收到消息失败后是否需要重新发送)* properties 属性信息* body 消息*/@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {//路由keylog.info("路由key为：" + envelope.getRoutingKey());//交换机log.info("交换机为：" + envelope.getExchange());//消息idlog.info("消息id为：" + envelope.getDeliveryTag());//收到的消息log.info("消费者2-接收到的消息为：" + new String(body, "utf8"));}};/*** 监听消息* 参数1：队列名称* 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认* 参数3：消息接收到后回调*/channel.basicConsume(QUEUE_RECEIVE, true, consumer);}
}

(5).Topics（主题模式/路由模式的一种）

生产者

/*** 跟 routing 路由模式类似，只不过路由模式是指定固定的路由键 routingKey，而主题模式是可以模糊匹配路由routingKey，类似于SQL中 = 和 like 的关系* 消息可能匹配多个消费者，但是同一个队列的中的消息不会被重复消费；**要求* Topic 模式消息的 routing_key 不能随意写，必须满足一定的要求，它必须是一个单词列表，以 “.” 或者 “#” 分隔开。这些单词可以是任意单词，这个单词列表最多不能超过 255 个字节。* 分隔符* “*(星号)”：可以代替一个单词* “#(井号)”：可以替代零个或多个单词* @Description:主题模式* @Author: xy丶*/
@Slf4j
public class TopicProducer {public static final String TOPIC_EXCHANGE = "topic_exchange";public static final String TOPIC_QUEUE_ONE = "topic_queue_one";public static final String TOPIC_QUEUE_TWO = "topic_queue_two";public static void main(String[] args) throws Exception {//声明用作全局变量的队列变量和交换价变量//创建连接Connection connection = RabbitMQConnection.getConnection();//创建信道Channel channel = connection.createChannel();//声明队列channel.queueDeclare(TOPIC_QUEUE_ONE,true,false,false,null);channel.queueDeclare(TOPIC_QUEUE_TWO,true,false,false,null);//声明交换机channel.exchangeDeclare(TOPIC_EXCHANGE, "topic",true);//绑定队列channel.queueBind(TOPIC_QUEUE_ONE,TOPIC_EXCHANGE,"*.orange.*");channel.queueBind(TOPIC_QUEUE_TWO,TOPIC_EXCHANGE,"*.*.rabbit");channel.queueBind(TOPIC_QUEUE_TWO,TOPIC_EXCHANGE,"lazy.#");//发生消息for (int i = 0; i <10 ; i++) {String msg="goodnight!My love world===>"+i;channel.basicPublish(TOPIC_EXCHANGE,"ag.we.rabbit",null,msg.getBytes());}}
}

消费者

@Slf4j
public class TopicCustomer1 {public static final String TOPIC_QUEUE_ONE="topic_queue_one";public static void main(String[] args) throws Exception{//创建连接Connection connection = RabbitMQConnection.getConnection();//创建信道Channel channel = connection.createChannel();DefaultConsumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {log.info("TopicCustomer1=====>:"+new String(body));}};channel.basicConsume(TOPIC_QUEUE_ONE,true,consumer);}
}@Slf4j
public class TopicCustomer2 {public static final String TOPIC_QUEUE_TWO="topic_queue_two";public static void main(String[] args) throws Exception{//创建连接Connection connection = RabbitMQConnection.getConnection();//创建信道Channel channel = connection.createChannel();DefaultConsumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {log.info("TopicCustomer22=====>:"+new String(body));}};channel.basicConsume(TOPIC_QUEUE_TWO,true,consumer);}
}@Slf4j
public class TopicCustomer3 {public static final String TOPIC_QUEUE_TWO = "topic_queue_two";public static void main(String[] args) throws Exception{//创建连接Connection connection = RabbitMQConnection.getConnection();//创建信道Channel channel = connection.createChannel();DefaultConsumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {log.info("TopicCustomer2=====>:"+new String(body));}};channel.basicConsume(TOPIC_QUEUE_TWO,true,consumer);}
}