RabbitMQ消息分发模式----"Hello World"模式

小鱼儿 2022-08-19 15:20 171阅读 0赞

"Hello World"模式：即最简单的分发模式：一个消息生产者，一个消息消费者。

![Center][]

生产者端代码：

public class MsgSender { 
 private final static String QUEUE_NAME = "hello"; 
 
 public static void main(String[] args) throws IOException { 
 /** 
 * 创建连接连接到RabbitMQ 
 */ 
 ConnectionFactory factory = new ConnectionFactory(); 
 // 设置MabbitMQ所在主机ip或者主机名 
 factory.setHost("127.0.0.1"); 
 // 创建一个连接 
 Connection connection = factory.newConnection(); 
 // 创建一个频道 
 Channel channel = connection.createChannel(); 
 // 指定一个队列 
 channel.queueDeclare(QUEUE_NAME, false, false, false, null); 
 // 发送的消息 
 String message = "hello world!"; 
 // 往队列中发出一条消息 
 channel.basicPublish("", QUEUE_NAME, null, message.getBytes()); 
 System.out.println(" [x] Sent '" + message + "'"); 
 // 关闭频道和连接 
 channel.close(); 
 connection.close(); 
 } 
 }

ConnectionFactory、Connection、Channel

ConnectionFactory、Connection、Channel，这三个都是RabbitMQ对外提供的API中最基本的对象。不管是服务器端还是客户端都会首先创建这三类对象。  
       ConnectionFactory为Connection的制造工厂。

Connection是与RabbitMQ服务器的socket链接，它封装了socket协议及身份验证相关部分逻辑。

Channel是我们与RabbitMQ打交道的最重要的一个接口，大部分的业务操作是在Channel这个接口中完成的，包括定义Queue、定义Exchange、绑定Queue与Exchange、发布消息等。

**Queue**

Queue（队列）是RabbitMQ的内部对象，用于存储消息，用下图表示。

![SouthEast][]

RabbitMQ中的消息都只能存储在Queue中，生产者（上图中的P---productor）生产消息并最终投递到Queue中，消费者（上图中的C---Consumer）可以从这个(hello)Queue（Hello）中获取消息并消费。

队列是由Channel声明的，而且这个操作是幂等的。同名的队列多次声明也只会创建一次。我们发送消息就是向这个声明的队列里发送消息。

消费者端代码：

public class MsgReceiver { 
 private final static String QUEUE_NAME = "hello"; 
 
 public static void main(String[] argv) throws IOException, InterruptedException { 
 ConnectionFactory factory = new ConnectionFactory(); 
 factory.setHost("127.0.0.1"); 
 // 打开连接和创建频道，与发送端一样 
 Connection connection = factory.newConnection(); 
 Channel channel = connection.createChannel(); 
 
 // 声明队列，主要为了防止消息接收者先运行此程序，队列还不存在时创建队列。 
 channel.queueDeclare(QUEUE_NAME, false, false, false, null); 
 // 创建队列消费者 
 QueueingConsumer consumer = new QueueingConsumer(channel); 
 System.out.println(" [*] Waiting for messages. To exit press CTRL+C"); 
 // 指定消费队列 true表示不用确认即可，只要接收，不管有没有成功处理都会当做成功 false只有当成功处理完毕并确认后才算
 channel.basicConsume(QUEUE_NAME, true, consumer); 
 while (true) { 
 // nextDelivery是一个阻塞方法（内部实现其实是阻塞队列的take方法） 
 QueueingConsumer.Delivery delivery = consumer.nextDelivery(); 
 String message = new String(delivery.getBody()); 
 System.out.println(" [x] Received '" + message + "'"); 
 // 返回确认状态 当channel.basicConsume()第二个参数为false时
             	 // 如果没有这一行确认接收则关闭消费者再次启动还会收到已经接收的消息。。。。
             // 为true时，如果加上这行报错channel error; protocol method: #method<channel.close>(reply-code=406,
             // reply-text=PRECONDITION_FAILED - unknown delivery tag 1,
             	 // class-id=60, method-id=80)
 // channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); 
 }
 }
 }

从消费者的代码中，我们可以看到ConnectionFactory、Connection、Channel这三个对象都还是会创建。而队列消费者这里再次声明一遍，是为了防止先启动消费者------当为消费者指定队列时，如果RabbitMQ服务器上未声明过该队列，就会抛出IO异常。

[http://blog.csdn.net/xiaoxian8023/article/details/48679609][http_blog.csdn.net_xiaoxian8023_article_details_48679609]

[Center]: /images/20220731/ddb689dc985449b6994bdb02b9cf52b8.png
[SouthEast]: /images/20220731/a71011e76f214e898a0f3f2ee6dfd518.png
[http_blog.csdn.net_xiaoxian8023_article_details_48679609]: http://blog.csdn.net/xiaoxian8023/article/details/48679609