概述

• 在上图的模型中，有以下概念
  • P：生产者，也就是要发送消息的程序
  • C：消费者，消息的接受者，会一直等待消息的到来
  • Queue：消息队列，图中蓝色部分
  • 类似一个邮箱，可以缓存消息
  • 生产者向其中投递消息，消费者从其中取出消息

创建生产者生产消息

• 代码如下：

/**
 * @author: BNZeng
 **/
public class Consumer {

    @Test
    public void receiveMessage() throws Exception {
        // 1.创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();

        // 2.设置连接相关信息的参数配置
        // RabbitMQ 服务器地址
        connectionFactory.setHost("192.168.0.130");
        // RabbitMQ 服务器数据交互端口号
        connectionFactory.setPort(5672);
        // RabbitMQ 服务器用户名
        connectionFactory.setUsername("user");
        // RabbitMQ 服务器密码
        connectionFactory.setPassword("123456");
        // RabbitMQ 服务器虚拟主机
        connectionFactory.setVirtualHost("v-it6666");

        // 3.从连接工厂里面创建一个连接
        Connection connection = connectionFactory.newConnection();

        // 4.创建通道
        Channel channel = connection.createChannel();

        // 5.绑定队列
        channel.queueDeclare("hello", true, false, false, null);

        // 6.接收消息
        channel.basicConsume("hello", true, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("get message success " + new String(body));
            }
        });

        // 7.不能让程序结束
        System.in.read();

        // 8.释放资源，关闭通道和连接
        channel.close();
        connection.close();
    }

}

• 这种就是一个 点对点 的发送和消费
• 一个生产者，一个消费者，可以用于登陆发送短信验证码等功能

抽取工具类

• 抽取之后的工具类代码如下：

/**
 * @author: BNZeng
 **/
public class Producer {

    @Test
    public void sendMessage() throws Exception {
        // 1.创建连接工厂
        // 2.设置连接相关信息的参数配置
        // 3.从连接工厂里面创建一个连接
        Connection connection = RabbitMQUtil.getConnection();

        // 4.创建通道
        Channel channel = connection.createChannel();

        // 5.绑定队列
        channel.queueDeclare("hello", true, false, false, null);

        // 6.发送消息
        channel.basicPublish("", "hello", null, "Hello RabbitMQ".getBytes());

        // 7.释放相关资源
        RabbitMQUtil.closeChannelAndConnection(channel, connection);

        System.out.println("message send success");
    }
}

修改消费者

• 修改之后的代码如下：

/**
 * @author: BNZeng
 **/
public class Consumer {

    @Test
    public void receiveMessage() throws Exception {
        // 1.创建连接工厂
        // 2.设置连接相关信息的参数配置
        // 3.从连接工厂里面创建一个连接
        Connection connection = RabbitMQUtil.getConnection();

        // 4.创建通道
        Channel channel = connection.createChannel();

        // 5.绑定队列
        channel.queueDeclare("hello", true, false, false, null);

        // 6.接收消息
        channel.basicConsume("hello", true, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("get message success " + new String(body));
            }
        });

        // 7.不能让程序结束
        System.in.read();

        // 8.释放相关资源
        RabbitMQUtil.closeChannelAndConnection(channel, connection);
    }

}

如果只是设置了队列的持久化，消息默认的是不会持久化的