RabbitMQ中自定义服务监听

Java光头强 • 2023年2月20日下午7:51 • 技术随笔 • 阅读 255

传入自定义的监听类，该类继承DefaultConsumer类，重写handleDelivery方法就行。

channel.basicConsume(quequName,false,new MyConsumer(channel));

生产端：

package com.wy.testrabbitmq.zidingyiconsumer;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

/**
 * @author wangyan@163.com
 * @version 1.0
 * @date 2019-06-12 17:20
 */
public class Producer {
    public static void main(String[] args) throws Exception {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setVirtualHost("/");
        connectionFactory.setHost("127.0.0.1");
        connectionFactory.setPort(5672);
        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();
        String exchangeName = "test_consumer_exchange";
        String routingkey = "test.consumer";
        String msg = "test consumer message";
        for (int i = 0; i < 7; i++) {
            channel.basicPublish(exchangeName, routingkey, true, null, msg.getBytes());
        }
    }
}

消费端：

package com.wy.testrabbitmq.zidingyiconsumer;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

/**
 * @author wangyan@163.com
 * @version 1.0
 * @date 2019-06-12 17:21
 */
public class Consumer {
    public static void main(String[] args)throws Exception {
        ConnectionFactory connectionFactory=new ConnectionFactory();
        connectionFactory.setVirtualHost("/");
        connectionFactory.setHost("127.0.0.1");
        connectionFactory.setPort(5672);
        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();
        String exchangeName="test_consumer_exchange";
        String routingkey="test.*";
        String quequName="test_consumer_queueName";
       channel.exchangeDeclare(exchangeName,"topic",true,false,null);
       channel.queueDeclare(quequName,true,false,false,null);
       channel.queueBind(quequName,exchangeName,routingkey);
        channel.basicConsume(quequName,true,new TestConsumer(channel));
    }
}

自定义类：

/**自定义消费端监听类
 * @author wangyan@163.com
 * @version 1.0
 * @date 2019-06-18 15:31
 */
public class TestConsumer extends DefaultConsumer {

    public TestConsumer(Channel channel) {
        super(channel);
    }
    @Override
    public void handleDelivery(String s, Envelope envelope, AMQP.BasicProperties basicProperties, byte[] body) throws IOException {
        System.out.println("s:"+s);
        System.out.println("envelope:"+envelope);
        System.out.println("basicProperties:"+basicProperties);
        System.out.println("body:"+new String(body));
    }
}