Native RabbitMQ Topic Exchange

RabbitMQ原生编程，Topic交换器。适用于平台给下游服务下发消息的业务场景，配合每个下游服务都有自己的vhost，实现消息隔离发送。

生产者：

/**
 * create by zhangjianbing
 * time 2020年9月1日
 */
public class TopicProducer {

    public final static String EXCHANGE_NAME = "topic_exchange";

    public static void main(String[] args) throws IOException, TimeoutException {
        // 创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("1.1.1.1");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("beijing");
        connectionFactory.setPassword("123456");
        connectionFactory.setVirtualHost("beijing");
        // 创建连接
        Connection connection = connectionFactory.newConnection();
        // 创建信道
        Channel channel = connection.createChannel();
        // 在信道中设置交换器
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC, true);

        String queueName1 = "ZOOBALL.CALLBACK.QUEUE";
        String queueName2 = "WALLET.CALLBACK.QUEUE";

        String routeKey1 = "LONGBALL.ROUTEKEY";
        String routeKey2 = "WALLET.ROUTEKEY";

        channel.queueDeclare(queueName1, true, false, false, null);
        channel.queueDeclare(queueName2, true, false, false, null);

        channel.queueBind(queueName1, EXCHANGE_NAME, routeKey1);
        channel.queueBind(queueName2, EXCHANGE_NAME, routeKey2);

        // 消息
        String message1 = "hello rabbit mq a";
        // 发送消息
        channel.basicPublish(EXCHANGE_NAME, routeKey1, null, message1.getBytes());

        // 消息
        String message2 = "hello rabbit mq b";
        // 发送消息
        channel.basicPublish(EXCHANGE_NAME, routeKey2, null, message2.getBytes());

        // 关闭信道和连接
        channel.close();
        connection.close();

    }

}

消费者：

/**
 * create by zhangjianbing
 * time 2020年9月1日
 */
@SuppressWarnings("Duplicates")
public class TopicConsumer {

    public static void main(String[] args) throws IOException, TimeoutException {
        // 创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("1.1.1.1");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("beijing");
        connectionFactory.setPassword("123456");
        connectionFactory.setVirtualHost("beijing");
        // 创建连接
        Connection connection = connectionFactory.newConnection();
        // 创建信道
        Channel channel = connection.createChannel();
        System.out.println("正在等待消息。。。。。。");
        // 声明一个消费者
        final Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                System.out.println("【路由键】：" + envelope.getRoutingKey() + "【消息内容】：" + new String(body, StandardCharsets.UTF_8));
            }
        };

        // 队列名称
        String queueName1 = "ZOOBALL.CALLBACK.QUEUE";
        // 消费者正式开始在指定队列上消费[队列名称、自动提交、消费者]
        channel.basicConsume(queueName1, true, consumer);

//        // 队列名称
//        String queueName2 = "WALLET.CALLBACK.QUEUE";
//        // 消费者正式开始在指定队列上消费[队列名称、自动提交、消费者]
//        channel.basicConsume(queueName2, true, consumer);


    }