消息中間件--RabbitMQ學習（十三）---進階特性之消費端自定義監聽

消費端自定義監聽

• 我們一般就是在代碼中編寫 while循環,進行 consumer.next Deliver方法進行擷取下一條消息,然後進行消費處理
• 但是我們使用自定義的 Consumere更加的友善,解耦性更加的強,也是在實際工作中最常用的使用方式

具體代碼實作

消費端代碼

public class Consumer {

    public static void main(String[] args) throws Exception{
        //1 建立一個connectionFactory
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.0.159");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        //2通過連接配接工場建立連接配接
        Connection connection = connectionFactory.newConnection();
        //3通過connection建立channel
        Channel channel = connection.createChannel();
        //開啟消息的确認模式
        channel.confirmSelect();
        String exchangeName = "test_consumer_exchange";
        String routingKey = "consumer.#";
        String queueName = "test_consumer_queue";
        channel.exchangeDeclare(exchangeName,"topic",true,false,null);
        channel.queueDeclare(queueName,true,false,false,null);
        channel.queueBind(queueName,exchangeName,routingKey);
        channel.basicConsume(queueName,true,new MyConsumer(channel));
    }
}
           

生産端代碼

public class Producter {
    public static void main(String[] args) throws Exception{
        //1 建立一個connectionFactory
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.0.159");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        //2通過連接配接工場建立連接配接
        Connection connection = connectionFactory.newConnection();
        //3通過connection建立channel
        Channel channel = connection.createChannel();
        //開啟消息的确認模式
        channel.confirmSelect();
        String exchangeName = "test_consumer_exchange";
        String routingKey = "consumer.save";

        //發送消息
        String msg = "hello";
        channel.basicPublish(exchangeName,routingKey,true,null,msg.getBytes());
    }
}
           

自定義消費者

public class MyConsumer extends DefaultConsumer {
    public MyConsumer(Channel channel) {
        super(channel);
    }

    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        System.err.println("consumerTag:  "+consumerTag);
        System.err.println("envelope  "+envelope);
        System.err.println("properties  "+properties);
        System.err.println("body  "+new String(body));
    }
}