RabbitMQ学习四 java客户端连接RabbitMQ

标签：connectionFactory java 队列 RabbitMQ 交换机参数消息 channel 客户端

RabbitMQ的工作模式，可以参考官网：https://www.rabbitmq.com/getstarted.html

一、简单方式

以下两种都是生产者直接发消息给队列，不通过交换机。且发送给队列的消息只能被消费一次。

比如一个队列被C1和C2消费，在队列中的消息只会被一个消费者消费。

生产者代码逻辑图

代码如下：

package com.java.coder.simple;

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

public class MyProducer {
    static final String QUEUE_NAME = "simple_queue";

    public static void main(String[] args) throws Exception {
        //创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        //主机地址;默认为 localhost
        connectionFactory.setHost("xxx.xx.x.xxx");
        //连接端口;默认为 5672
        connectionFactory.setPort(5672);
        //虚拟主机名称;默认为 /
        connectionFactory.setVirtualHost("/javacoder");
        //连接用户名；默认为guest
        connectionFactory.setUsername("xxx");
        //连接密码；默认为guest
        connectionFactory.setPassword("xxx");

        //创建连接
        Connection connection = connectionFactory.newConnection();

        // 创建频道
        Channel channel = connection.createChannel();

        // 声明（创建）队列,如果队列不存在则创建
        /**
         * 参数1：队列名称
         * 参数2：是否定义持久化队列
         * 参数3：是否独占本次连接
         * 参数4：是否在不使用的时候自动删除队列
         * 参数5：队列其它参数
         */
        channel.queueDeclare("log01", true, false, false, null);

        // 要发送的信息
        String message = "hello ,everyone";
        /**
         * 参数1：交换机名称，如果没有指定则使用默认Default Exchage
         * 参数2：路由key,简单模式下传递队列名称
         * 参数3：消息其它属性
         * 参数4：消息内容
         */
        channel.basicPublish("", "log01", null, message.getBytes());
        System.out.println("已发送消息：" + message);
        // 关闭资源
        channel.close();
        connection.close();
    }
}

消费者

消费者端代码也和生产者端差不多

package com.java.coder.simple;

import com.rabbitmq.client.*;

import java.io.IOException;

public class MyConsumer {
    public static void main(String[] args) throws Exception {
         //创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        //主机地址;默认为 localhost
        connectionFactory.setHost("xxx.xx.x.xxx");
        //连接端口;默认为 5672
        connectionFactory.setPort(5672);
        //虚拟主机名称;默认为 /
        connectionFactory.setVirtualHost("/javacoder");
        //连接用户名；默认为guest
        connectionFactory.setUsername("xxx");
        //连接密码；默认为guest
        connectionFactory.setPassword("xxx");

        //创建连接
        Connection connection = connectionFactory.newConnection();

        // 创建频道
        Channel channel = connection.createChannel();

        // 声明（创建）队列
        /**
         * 参数1：队列名称
         * 参数2：是否定义持久化队列
         * 参数3：是否独占本次连接
         * 参数4：是否在不使用的时候自动删除队列
         * 参数5：队列其它参数
         */
        channel.queueDeclare("log01", true, false, false, null);

        // 接收消息
        /**
         * queue 队列名称
         * autoAck 是否自动确认
         * callback 回调对象
         * basicConsume(String queue, boolean autoAck, Consumer callback)
         */
        Consumer defaultConsumer= new DefaultConsumer(channel) {
            /**
             * 回调方法，当接受到消息之后，会自动执行该方法
             * @param consumerTag 标识
             * @param envelope 获取一些信息，交换机、路由键等
             * @param properties 配置信息
             * @param body 消息数据
             * @throws IOException
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                //路由key
                System.out.println("路由key为：" + envelope.getRoutingKey());
                //交换机
                System.out.println("交换机为：" + envelope.getExchange());
                //消息id
                System.out.println("消息id为：" + envelope.getDeliveryTag());
                //收到的消息
                System.out.println("接收到的消息为：" + new String(body, "utf-8"));
            }
        };
        //监听消息
        /**
         * 参数1：队列名称
         * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认
         * 参数3：消息接收到后回调
         */
        channel.basicConsume("log01", true,defaultConsumer );
        // 消费者不关闭资源，应该一直监听消息
        //channel.close();
        //connection.close();

    }
}

二、fantout模式

订阅模式示例图：

前面2个案例中，只有3个角色：

P：生产者，也就是要发送消息的程序
C：消费者：消息的接受者，会一直等待消息到来。
queue：消息队列，图中红色部分

而在订阅模型中，多了一个exchange角色，而且过程略有变化：

P：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给X（交换机）
C：消费者，消息的接受者，会一直等待消息到来。
Queue：消息队列，接收消息、缓存消息。
Exchange：交换机，图中的X。一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。Exchange有常见以下3种类型：
- Fanout：广播，将消息交给所有绑定到交换机的队列
- Direct：定向，把消息交给符合指定routing key 的队列
- Topic：通配符，把消息交给符合routing pattern（路由模式）的队列

Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！

生产者代码

package com.java.coder.simple;

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

public class MyProducerFanout {

    //交换机名称
    static final String FANOUT_EXCHAGE = "fanout_exchange";

    public static void main(String[] args) throws Exception {
        //创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        //主机地址;默认为 localhost
        connectionFactory.setHost("xx.xx.xx.xxx");
        //连接端口;默认为 5672
        connectionFactory.setPort(5672);
        //虚拟主机名称;默认为 /
        connectionFactory.setVirtualHost("/javacoder");
        //连接用户名；默认为guest
        connectionFactory.setUsername("xxx");
        //连接密码；默认为guest
        connectionFactory.setPassword("xxx");

        //创建连接
        Connection connection = connectionFactory.newConnection();

        // 创建频道
        Channel channel = connection.createChannel();
        /**
         * 声明交换机
         * 参数1：交换机名称
         * 参数2：交换机类型，fanout、topic、direct、headers
         */
        channel.exchangeDeclare(FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT);

        // 声明（创建）队列
        /**
         * 参数1：队列名称
         * 参数2：是否定义持久化队列
         * 参数3：是否独占本次连接
         * 参数4：是否在不使用的时候自动删除队列
         * 参数5：队列其它参数
         */
        channel.queueDeclare("log01", true, false, false, null);
        channel.queueDeclare("log02", true, false, false, null);

        //队列绑定交换机
        channel.queueBind("log01", FANOUT_EXCHAGE, "");
        channel.queueBind("log02", FANOUT_EXCHAGE, "");
        // 要发送的信息
        for (int i = 1; i <= 10; i++) {
            // 发送信息
            String message = "你好；小兔子！发布订阅模式--" + i;
            /**
             * 参数1：交换机名称，如果没有指定则使用默认Default Exchage
             * 参数2：路由key,简单模式可以传递队列名称
             * 参数3：消息其它属性
             * 参数4：消息内容
             */
            channel.basicPublish(FANOUT_EXCHAGE, "", null, message.getBytes());
        }
        // 关闭资源
        channel.close();
        connection.close();
    }
}

进入RabbitMQ后台管理程序，可以看到队列中的消息，每个队列有10条消息

也可以看到，交换器fantou_exchange绑定了两个交换机

消费者

分别定义两个消费者，分别消费log01与log02队列中的消息

package com.java.coder.simple;

import com.rabbitmq.client.*;

import java.io.IOException;

public class MyConsumerFanout01 {

    
    public static void main(String[] args) throws Exception {
        //创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        //主机地址;默认为 localhost
        connectionFactory.setHost("101.43.3.119");
        //连接端口;默认为 5672
        connectionFactory.setPort(5672);
        //虚拟主机名称;默认为 /
        connectionFactory.setVirtualHost("/javacoder");
        //连接用户名；默认为guest
        connectionFactory.setUsername("java_coder");
        //连接密码；默认为guest
        connectionFactory.setPassword("java_coder");

        //创建连接
        Connection connection = connectionFactory.newConnection();

        // 创建频道
        Channel channel = connection.createChannel();

        // 声明（创建）队列 如果队列不存在则创建
        /**
         * 参数1：队列名称
         * 参数2：是否定义持久化队列
         * 参数3：是否独占本次连接
         * 参数4：是否在不使用的时候自动删除队列
         * 参数5：队列其它参数
         */
        channel.queueDeclare("log01", true, false, false, null);
        //队列绑定交换机
        channel.queueBind("log01", "fanout_exchange", "");

        // 接收消息
        /**
         * queue 队列名称
         * autoAck 是否自动确认
         * callback 回调对象
         * basicConsume(String queue, boolean autoAck, Consumer callback)
         */
        Consumer defaultConsumer= new DefaultConsumer(channel) {
            /**
             * 回调方法，当接受到消息之后，会自动执行该方法
             * @param consumerTag 标识
             * @param envelope 获取一些信息，交换机、路由键等
             * @param properties 配置信息
             * @param body 消息数据
             * @throws IOException
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                //路由key
                System.out.println("路由key为：" + envelope.getRoutingKey());
                //交换机
                System.out.println("交换机为：" + envelope.getExchange());
                //消息id
                System.out.println("消息id为：" + envelope.getDeliveryTag());
                //收到的消息
                System.out.println("接收到的消息为：" + new String(body, "utf-8"));
            }
        };
        //监听消息
        /**
         * 参数1：队列名称
         * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认
         * 参数3：消息接收到后回调
         */
        channel.basicConsume("log01", true,defaultConsumer );
        // 消费者不关闭资源，应该一直监听消息
        //channel.close();
        //connection.close();

    }
}

package com.java.coder.simple;

import com.rabbitmq.client.*;

import java.io.IOException;

public class MyConsumerFanout02 {


    public static void main(String[] args) throws Exception {
        //创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        //主机地址;默认为 localhost
        connectionFactory.setHost("101.43.3.119");
        //连接端口;默认为 5672
        connectionFactory.setPort(5672);
        //虚拟主机名称;默认为 /
        connectionFactory.setVirtualHost("/javacoder");
        //连接用户名；默认为guest
        connectionFactory.setUsername("java_coder");
        //连接密码；默认为guest
        connectionFactory.setPassword("java_coder");

        //创建连接
        Connection connection = connectionFactory.newConnection();

        // 创建频道
        Channel channel = connection.createChannel();

        // 声明（创建）队列 如果队列不存在则创建
        /**
         * 参数1：队列名称
         * 参数2：是否定义持久化队列
         * 参数3：是否独占本次连接
         * 参数4：是否在不使用的时候自动删除队列
         * 参数5：队列其它参数
         */
        channel.queueDeclare("log02", true, false, false, null);
        //队列绑定交换机
        channel.queueBind("log02", "fanout_exchange", "");

        // 接收消息
        /**
         * queue 队列名称
         * autoAck 是否自动确认
         * callback 回调对象
         * basicConsume(String queue, boolean autoAck, Consumer callback)
         */
        Consumer defaultConsumer= new DefaultConsumer(channel) {
            /**
             * 回调方法，当接受到消息之后，会自动执行该方法
             * @param consumerTag 标识
             * @param envelope 获取一些信息，交换机、路由键等
             * @param properties 配置信息
             * @param body 消息数据
             * @throws IOException
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                //路由key
                System.out.println("路由key为：" + envelope.getRoutingKey());
                //交换机
                System.out.println("交换机为：" + envelope.getExchange());
                //消息id
                System.out.println("消息id为：" + envelope.getDeliveryTag());
                //收到的消息
                System.out.println("接收到的消息为：" + new String(body, "utf-8"));
            }
        };
        //监听消息
        /**
         * 参数1：队列名称
         * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认
         * 参数3：消息接收到后回调
         */
        channel.basicConsume("log02", true,defaultConsumer );
        // 消费者不关闭资源，应该一直监听消息
        //channel.close();
        //connection.close();

    }
}

三、direct模式

路由模式特点：

队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）
消息的发送方在向 Exchange发送消息时，也必须指定消息的 RoutingKey。
Exchange不再把消息交给每一个绑定的队列，而是根据消息的Routing Key进行判断，只有队列的Routingkey与消息的 Routing key完全一致，才会接收到消息

图解：

P：生产者，向Exchange发送消息，发送消息时，会指定一个routing key。
X：Exchange（交换机），接收生产者的消息，然后把消息递交给与routing key完全匹配的队列
C1：消费者，其所在队列指定了需要routing key 为 error 的消息
C2：消费者，其所在队列指定了需要routing key 为 info、error、warning 的消息

生产者

package com.java.coder.simple;

import com.java.coder.constant.RabbitMqConstant;
import com.java.coder.util.ConnectionUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class DirectProducer {

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

        channel.queueDeclare(RabbitMqConstant.DIRECT_QUEUE_01, true, false, false, null);
        channel.queueDeclare(RabbitMqConstant.DIRECT_QUEUE_02, true, false, false, null);

        channel.exchangeDeclare(RabbitMqConstant.DIRECT_EXCHANGE, BuiltinExchangeType.DIRECT,true,false,null);

        channel.queueBind(RabbitMqConstant.DIRECT_QUEUE_01,RabbitMqConstant.DIRECT_EXCHANGE,RabbitMqConstant.DIRECT_ROUTING_INFO_KEY);
        channel.queueBind(RabbitMqConstant.DIRECT_QUEUE_02,RabbitMqConstant.DIRECT_EXCHANGE,RabbitMqConstant.DIRECT_ROUTING_ERROR_KEY);

        channel.basicPublish(RabbitMqConstant.DIRECT_EXCHANGE,RabbitMqConstant.DIRECT_ROUTING_INFO_KEY,null,"info 信息打印".getBytes());
        channel.basicPublish(RabbitMqConstant.DIRECT_EXCHANGE,RabbitMqConstant.DIRECT_ROUTING_ERROR_KEY,null,"error 信息打印".getBytes());
        channel.close();
        connection.close();

    }
}

消费者

package com.java.coder.simple;

import com.java.coder.constant.RabbitMqConstant;
import com.java.coder.util.ConnectionUtil;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

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

        // 接收消息
        /**
         * queue 队列名称
         * autoAck 是否自动确认
         * callback 回调对象
         * basicConsume(String queue, boolean autoAck, Consumer callback)
         */
        Consumer defaultConsumer= new DefaultConsumer(channel) {
            /**
             * 回调方法，当接受到消息之后，会自动执行该方法
             * @param consumerTag 标识
             * @param envelope 获取一些信息，交换机、路由键等
             * @param properties 配置信息
             * @param body 消息数据
             * @throws IOException
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                //路由key
                System.out.println("路由key为：" + envelope.getRoutingKey());
                //交换机
                System.out.println("交换机为：" + envelope.getExchange());
                //消息id
                System.out.println("消息id为：" + envelope.getDeliveryTag());
                //收到的消息
                System.out.println("接收到的消息为：" + new String(body, "utf-8"));
            }
        };
        //监听消息
        /**
         * 参数1：队列名称
         * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认
         * 参数3：消息接收到后回调
         */
        channel.basicConsume(RabbitMqConstant.DIRECT_QUEUE_01, true,defaultConsumer );
    }
}

四、topics模式

Topic类型与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符！

Routingkey一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert

通配符规则：

#：匹配一个或多个词

*：匹配不多不少恰好1个词

举例：

item.#：能够匹配`item.insert.abc 或者 item.insert
item.*：只能匹配item.insert

五、模式总结

RabbitMQ工作模式： 1、简单模式 HelloWorld 一个生产者、一个消费者，不需要设置交换机（使用默认的交换机）

2、工作队列模式 Work Queue 一个生产者、多个消费者（竞争关系），不需要设置交换机（使用默认的交换机）

3、发布订阅模式 Publish/subscribe 需要设置类型为fanout的交换机，并且交换机和队列进行绑定，当发送消息到交换机后，交换机会将消息发送到绑定的队列

4、路由模式 Routing 需要设置类型为direct的交换机，交换机和队列进行绑定，并且指定routing key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列

5、通配符模式 Topic 需要设置类型为topic的交换机，交换机和队列进行绑定，并且指定通配符方式的routing key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列

标签：connectionFactory,java,队列,RabbitMQ,交换机,参数,消息,channel,客户端
From： https://www.cnblogs.com/cplinux/p/17971393