rabbitMQ_helloworld

Posted 2020-09-21 zhenhong

在下图中，“P”是我们的生产者，“C”是我们的消费者。中间的框是队列 - RabbitMQ代表消费者的消息缓冲区。

 

 

本例使用maven构建项目，在pom.xml中添加一下依赖

<dependencies>
        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>4.1.0</version>
        </dependency>

        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>1.7.21</version>
        </dependency>

        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-simple</artifactId>
            <version>1.7.21</version>
        </dependency>

    </dependencies>

 

接下来通过生产者叫rabbitMQ服务器发送一个消息，这个消息被缓存在了rabbitMQ的队列中，生产者相当于发件人，rabbitMQ相当于申通快递，消费者相当于接收人

定义一个发件人Send.java

 1 package com.rabbitMQ;
 2 
 3 import com.rabbitmq.client.Channel;
 4 import com.rabbitmq.client.Connection;
 5 import com.rabbitmq.client.ConnectionFactory;
 6 
 7 /**
 8  * 发送消息后，如果服务器的队列中没有可发现的消费者，那么就会放在队列中等待消费者
 9  * 可以使用rabbitmqctl.bat list_queues命令查看某个队列的消息个数
10  * @author may
11  *
12  */
13 public class Send {
14 
15     private final static String QUEUE_NAME = "hello";
16 
17     public static void main(String[] args) throws Exception {
18         //创建连接工厂
19         ConnectionFactory factory = new ConnectionFactory();
20         //设置连接rabbitMQ服务器的ip
21         factory.setHost("localhost");
22 //        factory.setPort(5672);
23         //创建一个连接到服务器的链接
24         Connection connection = factory.newConnection();
25         //创建连接通道
26         Channel channel = connection.createChannel();
27         
28         /**
29          * 定义一个队列
30          * queueDeclare(String queue,//队列的名字
31          * boolean durable, //定义一个耐用队列，即持久化，如果RabbitMQ服务挂机，重启后还能恢复这个队列。
32          * boolean exclusive, //排他队列，只能在当前链接中可用，如果这个连接关闭，那么也就无效了。
33          * boolean autoDelete,//在连接断开后自动删除队列。
34          * Map<String, Object> arguments)
35          */
36         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
37         
38 //        channel.queueDeclareNoWait(queue, durable, exclusive, autoDelete, arguments);
39         
40         String message = "Hello World!";
41         /**
42          * Parameters:
43          *    exchange: the exchange to publish the message to 转发器的名字
44          *    routingKey: the routing key 路由key，这里就是队列的名字，表示要发送到这个队列上
45          *    props: other properties for the message - routing headers etc//这个信息的属性
46          *    body: the message body 要发送的信息
47          */
48         channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
49         
50         System.out.println(" [x] Sent \'" + message + "\'");
51         
52         channel.close();
53         
54         connection.close();
55     }
56 
57 }

 

接收消息，消费者从队列中获取消息

 

 

定义一个Recv.java

 

package com.rabbitMQ;

import java.io.IOException;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

public class Recv {
    //队列名要和发送者定义的队列一样的名字，它会告诉rabbitMQ它要再哪个队列中获取消息
    private final static String QUEUE_NAME = "hello";

    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        /**
         * queue the name of the queue
            durable true if we are declaring a durable queue (the queue will survive a server restart)
            exclusive true if we are declaring an exclusive queue (restricted to this connection)
            autoDelete true if we are declaring an autodelete queue (server will delete it when no longer in use)
            arguments other properties (construction arguments) for the queue
         */
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
        //定义一个消费者
        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                    byte[] body) throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println(" [x] Received \'" + message + "\'");
            }
        };
        //异步
        /**
         * queue the name of the queue 队列名
         * 如果autoAck设置为true，这个Consumer在收到消息之后会马上返回ack。服务器将立即删除在内存中排队的消息
         * false 在消息的任务处理完之后再手动ack，如果正在处理的时候，消费者发生异常，就不能返回ack，那么就不会删除这个消息，等待发现其他消费者发送给他们
            autoAck true if the server should consider messages acknowledged once delivered; false if the server should expect explicit acknowledgements
            callback an interface to the consumer object
         */
        channel.basicConsume(QUEUE_NAME, true, consumer);
        
        //rabbitmqctl.bat list_queues 命令可以列出当前有多少个队列
    }
}

 

运行Send，向rabbitMQ发送了一个消息，我们可以通过rabbitMQ的一个工具查看当前的队列中有多少个消息在排队。

hello是队列的名字，1表示当前的队列中存在一个消息等待着被消费者处理

这个时候，启动消费者去取快递，启动后输出一下内容，成功打印出了Hello World

[*] Waiting for messages. To exit press CTRL+C
[x] Received \'Hello World!\'

 

再次查看rabbitMQ中hello队列中的消息个数

消息个数为零。