介绍

:::tips image.png

基础的消息队列模型包括的三个角色：

  • publisher：消息发布者，将消息发送到队列queue
  • queue：消息队列，负责接受并缓存消息

  • consumer：订阅队列，处理队列中的消息 :::

    使用

    发布消息

    1. @SpringBootTest
    2. public class MyTest {
    4.   @Test
    5.   public void test() throws IOException, TimeoutException {
    6.       //构建连接工厂对象
    7.       ConnectionFactory factory = new ConnectionFactory();
    8.       //设置连接参数：主机名、端口号、vhost、用户名、密码
    9.       factory.setHost("RabbitMQ服务的IP");
    10.       factory.setPort(RabbitMQ服务的端口号);
    11.       factory.setVirtualHost("/");
    12.       factory.setUsername("用户名");
    13.       factory.setPassword("密码");
    15.       //创建连接对象
    16.       Connection connection = factory.newConnection();
    18.       //创建连接通道
    19.       Channel channel = connection.createChannel();
    21.       //创建队列
    22.       String queueName = "simple.queue";
    23.       channel.queueDeclare(queueName, false, false, false, null);
    25.       //发送消息
    26.       String message = "这是一条测试消息";
    27.       channel.basicPublish("", queueName, null, message.getBytes());
    28.       System.out.println("消息发送成功：" + message);
    30.       //关闭连接
    31.       channel.close();
    32.       connection.close();
    33.   }
    34. }

    订阅消息

    1. @SpringBootTest
    2. public class MyTest {
    4.   public void test() throws IOException, TimeoutException {
    5.       //构建连接工厂对象
    6.       ConnectionFactory factory = new ConnectionFactory();
    7.       //设置连接参数：主机名、端口号、vhost、用户名、密码
    8.       factory.setHost("RabbitMQ服务的IP");
    9.       factory.setPort(RabbitMQ服务的端口号);
    10.       factory.setVirtualHost("/");
    11.       factory.setUsername("用户名");
    12.       factory.setPassword("密码");
    13.       //创建连接对象
    14.       Connection connection = factory.newConnection();
    16.       //创建连接通道
    17.       Channel channel = connection.createChannel();
    19.       //创建队列
    20.       String queueName = "simple.queue";
    21.       channel.queueDeclare(queueName, false, false, false, null);
    23.       //订阅消息
    24.       channel.basicConsume(queueName, true, new DefaultConsumer(channel){
    25.           @Override
    26.           public void handleDelivery(String consumerTag, Envelope envelope,
    27.                                      AMQP.BasicProperties properties, byte[] body) throws IOException {
    28.               //处理消息
    29.               String message = new String(body);
    30.               System.out.println("接收到消息：" + message);
    31.           }
    32.       });
    33.       System.out.println("等待接收消息");
    34.   }
    35. }