RabbitMQ入门教学

作者: 指针v587 | 来源:发表于2018-12-27 16:32 被阅读0次

MyBatis-从查询昨天的数据说起
RabbitMQ入门教学
RabbitMQ 入门教学（一）
消息服务 - RabbitMQ 基础入门
RabbitMQ基础
RabbitMQ入门-高效的Work模式
RabbitMQ 入门
RabbitMQ入门-从HelloWorld开始
RabbitMq相关文章索引（1）
RabbitMQ学习

一. 背景

　　拿到代码以后，发现，生产者在向外发送消息时，指定了exchange（交换机）和routing key，但是没有指定queue（队列）也没有将queue（队列）绑定到exchange，刚开始因为不熟悉rabbitMQ，所有不知道怎么回事，后来知道了：消费者在消费消息时，需要声明队列（队列名字随便），并将声明的队列通过routing key绑定到exchange，这样才能接收到数据，因此，生产者方需要将exchange和routing key实现告知消费者方。

二. 代码实例：生产者方指定了exchange（交换机）和routing key，但是不指定queue（队列）也不将queue（队列）绑定到exchange，队列声明和绑定队列到exchange的工作由消费者方完成

　　1. 生产者方

　　　　① 生产者方代码

1import java.io.IOException; 2import com.rabbitmq.client.Channel; 3import com.rabbitmq.client.Connection; 4import com.rabbitmq.client.ConnectionFactory; 5 6publicclass Producer { 7privatefinalstaticString QUEUE_NAME = "QUEUE8"; 8 9publicstaticvoidmain(String[] args)throws IOException { 1011ConnectionFactory factory =new ConnectionFactory(); 12factory.setHost("localhost");13factory.setPort(5672);14factory.setUsername("guest");15factory.setPassword("guest");1617Connection connection = factory.newConnection(); 18Channel channel = connection.createChannel(); 1920String message = "Hello World!"; 2122// 指定exchange和routing key，并发送消息到exchange23channel.basicPublish("FILETOPIC", "KEY.FILE",null, message.getBytes()); 24System.out.println(" [x] Sent '" + message + "'"); 2526 channel.close(); 27 connection.close(); 28 } 29}

　　　　② 生产者方代码运行后，可在rabbiteMQ managerment 管理界面看到相应exchange，如下图所示：

---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------

--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------

　　2. 消费者方：消费者声明队列，队列名称随便起，并将该队列通过生产者指定的routing key绑定的其指定的exchange上

　　　　① 消费者方代码

1import com.rabbitmq.client.ConnectionFactory; 2import com.rabbitmq.client.QueueingConsumer; 3import com.rabbitmq.client.Channel; 4import com.rabbitmq.client.Connection; 5 6publicclass Reqv { 7 8privatefinalstaticString QUEUE_NAME = "QUEUE8"; 910publicstaticvoidmain(String[] argv)throws Exception { 1112ConnectionFactory factory =new ConnectionFactory();13factory.setPort(5672); 14factory.setUsername("guest");15factory.setPassword("guest");16factory.setHost("localhost");1718Connection connection = factory.newConnection(); 19Channel channel = connection.createChannel(); 2021// 声明队列22channel.queueDeclare(QUEUE_NAME,false,false,false,null); 23System.out.println(" [*] Waiting for messages. To exit press CTRL+C");2425// 绑定队列到交换机26channel.queueBind(QUEUE_NAME, "FILETOPIC", "KEY.FILE");2728QueueingConsumer consumer =new QueueingConsumer(channel); 29channel.basicConsume(QUEUE_NAME,true, consumer); 30while(true){ 31QueueingConsumer.Delivery delivery = consumer.nextDelivery(); 32String message =newString(delivery.getBody(),"UTF-8"); 33System.out.println(" 【[x] Received 】：" + message); 34 } 35 } 36}

　　　　② 运行效果，消费者方代码运行后，在rabbiteMQ managerment 管理界面可以看到声明的队列，并发现该队列已经绑定到了生产者指定的exchange上

---------------------------------------------------------------------------------------------------------------------------------------------------------------

--------------------------------------------------------------------------------------------------------------------------------------------------------------------

---------------------------------------------------------------------------------------------------------------------------------------------------------------

三. 代码实例：生产者指定exchange和routing key，声明队列并将队列绑定到exchange，消费者只需从生产者绑定的队列消费即可

　　1. 生产者

　　　　① 生产者代码

1import java.io.IOException; 2import com.rabbitmq.client.Channel; 3import com.rabbitmq.client.Connection; 4import com.rabbitmq.client.ConnectionFactory; 5 6publicclass Producer { 7privatefinalstaticString QUEUE_NAME = "QUEUE1"; 8 9publicstaticvoidmain(String[] args)throws IOException { 1011ConnectionFactory factory =new ConnectionFactory(); 12factory.setHost("localhost");13factory.setPort(5672);14factory.setUsername("guest");15factory.setPassword("guest");1617Connection connection = factory.newConnection(); 18Channel channel = connection.createChannel(); 1920String message = "Hello World!"; 2122// 声明队列23channel.queueDeclare(QUEUE_NAME,true,false,false,null);2425// 绑定队列到交换机26channel.queueBind(QUEUE_NAME, "FILETOPIC", "KEY.FILE");2728// 指定exchange和routing key，并发送消息到exchange29channel.basicPublish("FILETOPIC", "KEY.FILE",null, message.getBytes()); 30System.out.println(" [x] Sent '" + message + "'"); 3132 channel.close(); 33 connection.close(); 34 } 35}

　　　　② 生产者运行效果：生产者代码运行后，在rabbiteMQ managerment 管理界面可以看到生产者声明的队列，并发现该队列已经绑定到了生产者指定的exchange上，如下图所示：

　　2. 消费者

　　　　① 消费者代码：消费者现在直接消费队列中的消息即可，既不用声明队列，也不用绑定

1import com.rabbitmq.client.ConnectionFactory; 2import com.rabbitmq.client.QueueingConsumer; 3import com.rabbitmq.client.Channel; 4import com.rabbitmq.client.Connection; 5 6publicclass Reqv { 7 8privatefinalstaticString QUEUE_NAME = "QUEUE1"; 910publicstaticvoidmain(String[] argv)throws Exception { 1112ConnectionFactory factory =new ConnectionFactory();13factory.setPort(5672); 14factory.setUsername("guest");15factory.setPassword("guest");16factory.setHost("localhost");1718Connection connection = factory.newConnection(); 19Channel channel = connection.createChannel(); 2021QueueingConsumer consumer =new QueueingConsumer(channel); 22channel.basicConsume(QUEUE_NAME,true, consumer); 23while(true){ 24QueueingConsumer.Delivery delivery = consumer.nextDelivery(); 25String message =newString(delivery.getBody(),"UTF-8"); 26System.out.println(" 【[x] Received 】：" + message); 27 } 28 } 29}