Java测试RabbitMQ生产者和消费者

刺骨的言语ヽ痛彻心扉 2022-05-21 22:41 169阅读 0赞

# 一、环境准备 #

请看上一篇文章，[Ubuntu18.04安装RabbitMQ][Ubuntu18.04_RabbitMQ]完成RabbitMQ服务端搭建，并创建admin用户。

gradle添加RabbitMQ客户端依赖。

compile group: 'com.rabbitmq', name: 'amqp-client', version: '3.6.5'

# 二、生产者创建 #

用户名和密码都为admin，队列名字为test

import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /**
     * Created by wzj on 2018/6/10.
     */
    public class Producer
    {
        public final static String QUEUE_NAME = "test";
    
        public static void main(String[] args) throws IOException, TimeoutException
        {
            //创建连接工厂
            ConnectionFactory factory = new ConnectionFactory();
    
            //设置RabbitMQ相关信息
            factory.setHost("192.168.3.45");
            factory.setVirtualHost("/");
            factory.setUsername("admin");
            factory.setPassword("admin");
            factory.setPort(5672);
    
            //创建一个新的连接
            Connection connection = factory.newConnection();
    
            //创建一个通道
            Channel channel = connection.createChannel();
    
            // 声明一个队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            //发送消息到队列中
            String message = "Hello RabbitMQ";
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));
            System.out.println("Producer Send +'" + message + "'");
    
            //关闭通道和连接
            channel.close();
            connection.close();
        }
    }

# 三、消费者创建 #

import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /**
     * Created by wzj on 2018/6/10.
     */
    public class Customer
    {
        private final static String QUEUE_NAME = "test";
    
        public static void main(String[] args) throws IOException, TimeoutException
        {
            // 创建连接工厂
            ConnectionFactory factory = new ConnectionFactory();
    
            //设置RabbitMQ地址
            factory.setHost("192.168.3.45");
            factory.setVirtualHost("/");
            factory.setUsername("admin");
            factory.setPassword("admin");
    
            //创建一个新的连接
            Connection connection = factory.newConnection();
    
            //创建一个通道
            Channel channel = connection.createChannel();
    
            //声明要关注的队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            System.out.println("Customer Waiting Received messages");
    
            // 告诉服务器我们需要那个频道的消息，如果频道中有消息，就会执行回调函数handleDelivery
            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("Customer Received '" + message + "'");
                }
            };
            //自动回复队列应答 -- RabbitMQ中的消息确认机制
            channel.basicConsume(QUEUE_NAME, true, consumer);
        }
    }

# 四、测试 #

1、启动消费者程序，程序会阻塞，等待生成者发送消息。

2、启动生产者程序，会发现消费者程序接收到了消息。

Customer Waiting Received messages
    Customer Received 'Hello RabbitMQ'

[Ubuntu18.04_RabbitMQ]: https://blog.csdn.net/u010889616/article/details/80643892