RabbitMq-Demo-05-主题模式

RabbitMq-Demo-05-主题模式 1、代码编写

1、生产者代码编写：

public class Producer_Topics {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1、获取连接
        ConnUtils cs = new ConnUtils();
        Connection connection = cs.getConnection();

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

        // 3、创建交换机
        String exchangeName = "test_topic";
        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC,true,false,false,null);

        // 4、创建队列
        String queue1Name = "test_topic_queue1";
        String queue2Name = "test_topic_queue2";
        channel.queueDeclare(queue1Name,true,false,false,null);
        channel.queueDeclare(queue2Name,true,false,false,null);

        // 5、绑定交换机和队列
        channel.queueBind(queue1Name,exchangeName,"#.t1");
        channel.queueBind(queue1Name,exchangeName,"t1.*");
        channel.queueBind(queue2Name,exchangeName,"*.*");

        // 6、发送消息
        String body = "test_topic";
        channel.basicPublish(exchangeName,"q1.t1",null,body.getBytes());

        // 7、关闭链接
        channel.close();
        connection.close();
    }
}

2、消费者01代码编写

public class Consumer_Topics1 {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1、获取连接
        ConnUtils cs = new ConnUtils();
        Connection connection = cs. getConnection();

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

        // 3、接受消息
        String queue1Name = "test_topic_queue1";
        Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("body：" + new String(body));
            }
        };
        // 4、监听消息
        channel.basicConsume(queue1Name,true,consumer);
    }
}

3、消费者02代码编写：

public class Consumer_Topics2 {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1、获取连接
        ConnUtils cs = new ConnUtils();
        Connection connection = cs. getConnection();

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

        // 3、接受消息
        String queue2Name = "test_topic_queue2";
        Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("body："+new String(body));
            }
        };
        // 4、监听消息
        channel.basicConsume(queue2Name,true,consumer);
    }
}

2、测试

1、运行生产者代码
2、查看消费者01控制带
3、查看消费者02控制台

均有消息：
	body：test_topic

项目代码链接：https://github.com/Mbm7280/rabbitmq_demo