rabbitmq java客户端如何使用

2022-11-12 12:16:18

rabbitmq java客户端如何使用

增加依赖

  <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>5.1.2</version>
        </dependency>

模型

目前官网提供7重模型我们就用java实现

hello

rabbitmq java客户端如何使用
p: 消息生成者
c: 消息消费者
红色:消息队列,进行消息存放和缓存的地方
生成者:

public static void main(String[] args) throws IOException, TimeoutException {
        //创建一个连接工厂对象
        ConnectionFactory connectionFactory = new ConnectionFactory();
        //设置连接主机
        connectionFactory.setHost("ip");
        //设置端口
        connectionFactory.setPort(5672);
        //设置虚拟主机
        connectionFactory.setVirtualHost("test");
        connectionFactory.setPassword("test");
        connectionFactory.setUsername("test");

        //获取连接对象
        Connection connection = connectionFactory.newConnection();
        //获取连接中通道
        Channel channel = connection.createChannel();

        channel.queueDeclare("hello",false,false,false,null);

        channel.basicPublish("","hello",null,"hello rabbitmq".getBytes());

        channel.close();
        connection.close();


    }

消费者

 public static void main(String[] args) throws IOException, TimeoutException {
        //创建一个连接工厂对象
        ConnectionFactory connectionFactory = new ConnectionFactory();
        //设置连接主机
        connectionFactory.setHost("ip");
        //设置端口
        connectionFactory.setPort(5672);
        //设置虚拟主机
        connectionFactory.setVirtualHost("test");
        connectionFactory.setPassword("test");
        connectionFactory.setUsername("test");

        //获取连接对象
        Connection connection = connectionFactory.newConnection();
        //获取连接中通道
        Channel channel = connection.createChannel();
         /**
         * 信道绑定队列
         * 参数1: 队列名称
         * 参数2:是否持久化
         * 参数3:是否独占队列
         * 参数4:使用完成是否自动删除队列
         * 参数5:额外参数附近配置
         */
        channel.queueDeclare("hello",false,false,false,null);

        /**
         * 发布消息
         * 参数1:交换机名称
         * 参数2:队列名称
         * 参数3:传递消息的额外配置
         * 参数4:具体的消息
         */
        channel.basicConsume("hello",true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

                System.out.println(new String(body) );
            }
        });
    }

工具的封装

public static Connection getConnection() throws IOException, TimeoutException {
        //创建一个连接工厂对象
        ConnectionFactory connectionFactory = new ConnectionFactory();
        //设置连接主机
        connectionFactory.setHost("ip");
        //设置端口
        connectionFactory.setPort(5672);
        //设置虚拟主机
        connectionFactory.setVirtualHost("test");
        connectionFactory.setPassword("test");
        connectionFactory.setUsername("test");

        //获取连接对象
        Connection connection = connectionFactory.newConnection();
        return connection;
    }

    public static void close(Connection connection, Channel channel) throws IOException, TimeoutException {
        if(connection!=null) {connection.close();}
        if (channel!=null) {channel.close();}

    }

ConnectionFactory 创建工厂是很重的,我们不能使用一次创建一次放到静态块里面吧还是

public class MQUtil {

    private static ConnectionFactory connectionFactory;

    static {
        //创建一个连接工厂对象
        connectionFactory = new ConnectionFactory();
        //设置连接主机
        connectionFactory.setHost("ip");
        //设置端口
        connectionFactory.setPort(5672);
        //设置虚拟主机
        connectionFactory.setVirtualHost("test");
        connectionFactory.setPassword("test");
        connectionFactory.setUsername("test");
    }
    public static Connection getConnection() throws IOException, TimeoutException {


        //获取连接对象
        Connection connection = connectionFactory.newConnection();
        return connection;
    }

    public static void close(Connection connection, Channel channel) throws IOException, TimeoutException {
        if(connection!=null) {connection.close();}
        if (channel!=null) {channel.close();}

    }
}

work

rabbitmq java客户端如何使用
从上图可以看到一个消息生产者,多个消息消费者:目的就是解决当生产者生产消息比较快,消费者消费比较慢,这样消息越积越多,所以需要多个消费者进行消费,work模型:将多个消费者绑定消息队列,消息一旦消费就会消失,所以也不会出现一个消息被多个消费者进行消费

生产者:

public class Provider {

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = MQUtil.getConnection();

        Channel channel = connection.createChannel();

        channel.queueDeclare("work",false,false,false,null);

        for (int i = 0; i <10 ; i++) {
            channel.basicPublish("","work",null,(i+"work 队列").getBytes());
        }

        MQUtil.close(connection,channel);

    }
}

消费者1:

public class Consumer1 {
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = MQUtil.getConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare("work",false,false,false,null);

        channel.basicConsume("work", true, new DefaultConsumer(channel) {

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

    }
}

消费者2:

public class Consumer2 {
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = MQUtil.getConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare("work",false,false,false,null);

        channel.basicConsume("work",true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者2: ="+ new String(body));
            }
        });
    }
}

rabbitmq java客户端如何使用
rabbitmq java客户端如何使用

这么一看是平均主义,如果有的能力强,有的能力弱,一定存在累死的和闲死的
不妨把消费者2睡眠3s,看看效果

能者多劳

Doing a task can take a few seconds. You may wonder what happens if one of the consumers starts a long task and dies with it only partly done. With our current code, once RabbitMQ delivers a message to the consumer it immediately marks it for deletion. In this case, if you kill a worker we will lose the message it was just processing. We’ll also lose all the messages that were dispatched to this particular worker but were not yet handled.

public class Consumer2 {
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = MQUtil.getConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare("work",false,false,false,null);
        //每次只能消费一个消息
        channel.basicQos(1);
        //关闭自动确认
        channel.basicConsume("work",false,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("消费者2: ="+ new String(body));
                /**
                 * 消息确认
                 * 参数1: 指定具体消费哪个消息
                 * 参数2: 是否同时确认多个消息
                 */
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        });
    }
}

//每次只能消费一个消息
channel.basicQos(1);
//关闭自动确认
channel.basicConsume(“work”,false,new DefaultConsumer(channel)

/**
* 消息确认
* 参数1: 指定具体消费哪个消息
* 参数2: 是否同时确认多个消息
*/
channel.basicAck(envelope.getDeliveryTag(),false);

fanout 广播

publish/subscribe

rabbitmq java客户端如何使用

  • 可以有多个消费者
  • 每个消费者有自己的quene队列
  • 每个队列都要绑定到Exchange(交换机)
  • 生成者发送的消息,只能发送到交换机,交换机来决定要发个哪个队列,生产者无法决定
  • 队列的消费者都能拿到消息,实现一条消息被多个消费者消费

注意: 消费者需要先启动,如果没有启动的消费者不会消费之前的消息

生产者:

public class Provider {
    private static final String EXCHANGE_NAME = "logs";
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = MQUtil.getConnection();
        Channel channel = connection.createChannel();
        //声明交换机
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
        channel.basicPublish(EXCHANGE_NAME, "", null, "fanout type message".getBytes("UTF-8"));
    }
}

消费者1:

public class Consumer1 {
    private static final String EXCHANGE_NAME = "logs";
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = MQUtil.getConnection();
        Channel channel = connection.createChannel();
        //绑定交换机
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
        //获取队列
        String queueName = channel.queueDeclare().getQueue();
        //fanout 类型的交换机 routingKey 没多大意义
        channel.queueBind(queueName, EXCHANGE_NAME, "");

        channel.basicConsume(queueName,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者1: =="+new String(body));
            }
        });
    }
}

消费者2:

public class Consumer2 {
    private static final String EXCHANGE_NAME = "logs";
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = MQUtil.getConnection();
        Channel channel = connection.createChannel();
        //绑定交换机
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
        //获取队列
        String queueName = channel.queueDeclare().getQueue();
        //fanout 类型的交换机 routingKey 没多大意义
        channel.queueBind(queueName, EXCHANGE_NAME, "");

        channel.basicConsume(queueName,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者2: =="+new String(body));
            }
        });
    }
}

rabbitmq java客户端如何使用
rabbitmq java客户端如何使用

direct

rabbitmq java客户端如何使用
rabbitmq java客户端如何使用

fanout模型我们知道,交换机将消息给所有绑定的队列,如果我们想不同的消息被不同的队列进行消费就需要使用direct模型

  • 队列和交换机不能任意绑定,需要指定一个rountingkey
  • 消息向交换机发送消息时也必须指定rountingkey
  • 交换机不再把消息推送个所有绑定的队列了,而是根据消息的rountingkey和队列的rounting可以完全一致,才进行推送(订阅模式)

生产者

public class Provider {
    private static final String EXCHANGE_NAME = "direct_logs";
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = MQUtil.getConnection();
        Channel channel = connection.createChannel();

        channel.exchangeDeclare(EXCHANGE_NAME,"direct");

        channel.basicPublish(EXCHANGE_NAME,"info",null,"routingKey info ".getBytes());

        channel.basicPublish(EXCHANGE_NAME,"error",null,"routingKey error ".getBytes());


    }
}

消费者ConsumerError1 :

public class ConsumerError1 {
    private static final String EXCHANGE_NAME = "direct_logs";
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = MQUtil.getConnection();
        Channel channel = connection.createChannel();
        //绑定交换机
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");
        //获取队列
        String queueName = channel.queueDeclare().getQueue();
       
        channel.queueBind(queueName, EXCHANGE_NAME, "error");

        channel.basicConsume(queueName,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者1error: =="+new String(body));
            }
        });
    }
}

消费者2ConsumerError2

public class ConsumerError2 {
    private static final String EXCHANGE_NAME = "direct_logs";
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = MQUtil.getConnection();
        Channel channel = connection.createChannel();
        //绑定交换机
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");
        //获取队列
        String queueName = channel.queueDeclare().getQueue();
        //fanout 类型的交换机 routingKey 没多大意义
        channel.queueBind(queueName, EXCHANGE_NAME, "error");

        channel.basicConsume(queueName,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者error2: =="+new String(body));
            }
        });
    }
}

消费者3 ConsumerInfo1

public class ConsumerInfo1 {
    private static final String EXCHANGE_NAME = "direct_logs";
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = MQUtil.getConnection();
        Channel channel = connection.createChannel();
        //绑定交换机
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");
        //获取队列
        String queueName = channel.queueDeclare().getQueue();
        //fanout 类型的交换机 routingKey 没多大意义
        channel.queueBind(queueName, EXCHANGE_NAME, "info");

        channel.basicConsume(queueName,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者info1: =="+new String(body));
            }
        });
    }
}

消费者4 ConsumerInfo2

public class ConsumerInfo2 {
    private static final String EXCHANGE_NAME = "direct_logs";
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = MQUtil.getConnection();
        Channel channel = connection.createChannel();
        //绑定交换机
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");
        //获取队列
        String queueName = channel.queueDeclare().getQueue();
        //fanout 类型的交换机 routingKey 没多大意义
        channel.queueBind(queueName, EXCHANGE_NAME, "info");

        channel.basicConsume(queueName,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者info2: =="+new String(body));
            }
        });
    }
}

topic

topic和direct差不多,路由键可以是用通配符来进行匹配

rabbitmq java客户端如何使用

*(star) can substitute for exactly one word.
#(hash) can substitute for zero or more words.

*匹配一个字符
#匹配0个或者多个字符

生产者:

public class TopicProducer {
    public static final String EXCHANENAME="topic_logs";
    public static void main(String[] args) throws IOException, TimeoutException {

        Channel channel = MQUtils.getChannel();
        //声明交换机
        channel.exchangeDeclare(EXCHANENAME,"topic");

        channel.basicPublish(EXCHANENAME,"log.info.leve2",null, "log.info.leve2".getBytes());
        channel.basicPublish(EXCHANENAME,"log.info.leve1",null, "log.info.leve1".getBytes());
        channel.basicPublish(EXCHANENAME,"log.info.leve1.aa.ee.cc",null, "log.info.leve1.aa.ee.cc".getBytes());

    }
}

消费者1:

public class TopicConstomer1 {

    public static void main(String[] args) throws IOException, TimeoutException {

        Channel channel = MQUtils.getChannel();

        channel.exchangeDeclare(TopicProducer.EXCHANENAME,"topic");

        String queue = channel.queueDeclare().getQueue();
        System.out.println("消费者1 队列  "+ queue);
        channel.queueBind(queue, TopicProducer.EXCHANENAME,"log.info.*");

        channel.basicConsume(queue,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者1 log.info.*: =="+new String(body));
            }
        });
    }
}

rabbitmq java客户端如何使用
消费者2:

public class TopicConstomer2 {

    public static void main(String[] args) throws IOException, TimeoutException {

        Channel channel = MQUtils.getChannel();

        channel.exchangeDeclare(TopicProducer.EXCHANENAME,"topic");

        String queue = channel.queueDeclare().getQueue();
        System.out.println("消费者2 队列  "+ queue);
        channel.queueBind(queue, TopicProducer.EXCHANENAME,"log.*.*");

        channel.basicConsume(queue,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者2 log.*.*: =="+new String(body));
            }
        });
    }
}

rabbitmq java客户端如何使用
消费者3:

public class TopicConstomer3 {

    public static void main(String[] args) throws IOException, TimeoutException {

        Channel channel = MQUtils.getChannel();

        channel.exchangeDeclare(TopicProducer.EXCHANENAME,"topic");

        String queue = channel.queueDeclare().getQueue();
        System.out.println("消费者3 队列  "+ queue);
        channel.queueBind(queue, TopicProducer.EXCHANENAME,"log.#");

        channel.basicConsume(queue,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者3 log.#: =="+new String(body));
            }
        });
    }
}

rabbitmq java客户端如何使用
消费者4:

public class TopicConstomer4 {

    public static void main(String[] args) throws IOException, TimeoutException {

        Channel channel = MQUtils.getChannel();

        channel.exchangeDeclare(TopicProducer.EXCHANENAME,"topic");

        String queue = channel.queueDeclare().getQueue();
        System.out.println("消费者4 队列  "+ queue);
        channel.queueBind(queue, TopicProducer.EXCHANENAME,"*.info.#");

        channel.basicConsume(queue,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者4 *.info.*: =="+new String(body));
            }
        });
    }
}

rabbitmq java客户端如何使用

上一篇：day018客户端服务端传输


下一篇：Java语言程序设计（十五）通过Java语言读取文本文件.txt