Java使用RabbitMQ之订阅分发（Topic）

        使用RabbitMQ进行消息发布和订阅，生产者将消息发送给转发器（exchange），转发器根据路由键匹配已绑定的消息队列并转发消息，主题模式支持路由键的通配。

生产者代码：

package org.study.exchange3.topic3;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import org.junit.Test;
import org.study.utils.ConnectionUtils;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * topic-主题模式（分发订阅）
 * exchange只转发消息，但是没有存储能力，只有队列才有存储能力
 * 主题模式支持路由键的通配符
 * “#”表示0个或若干个关键字，“*”表示一个关键字。
 */
public class Sender {
    public static final String QUEUE_NAME = "test_topic_queue";
    public static final String EXCHANGE_NAME = "topic_exchange";

    @Test
    public void send() throws IOException, TimeoutException, InterruptedException {
        // 获取连接
        Connection conn = ConnectionUtils.getConnection();
        // 获取通道
        Channel channel = conn.createChannel();
//        //创建队列
//        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        //声明转发器
        channel.exchangeDeclare(EXCHANGE_NAME, "topic");
        //每个消费者发送确认消息前，只发送一条消息
        channel.basicQos(1);

        String msg = "hello rabbitmq topic !";
        //发送消息至转发器,指定路由键
        channel.basicPublish(EXCHANGE_NAME, "key.key", null, msg.getBytes());
        System.out.println("[send] msg " + msg);

        channel.close();
        conn.close();
    }
}

消费者代码：

package org.study.exchange3.topic3;

import com.rabbitmq.client.*;
import org.junit.Test;
import org.study.utils.ConnectionUtils;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * 主题模式-接收消息
 */
public class Recv {
    public static final String QUEUE_NAME = "test_topic_queue";
    public static final String EXCHANGE_NAME = "topic_exchange";

    @Test
    public void recv() throws IOException, TimeoutException, InterruptedException {
        Connection conn = ConnectionUtils.getConnection();
        Channel channel = conn.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        channel.basicQos(1);
        /*
         * 队列绑定转发器,路由键通配符#和*
         * #：表示0个或多个字符
         * *：表示一个字符
         * */
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "key.#");

        //定义消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            //重写获取到达消息
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body, "utf-8");
                System.out.println("[1] recv: " + msg);

                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    System.out.println("[1] done!");
                    // 回执
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        };

        while (true) {
            //监听队列
            channel.basicConsume(QUEUE_NAME, false, consumer);
            Thread.sleep(1000);
        }

    }
}