RabbitMQ常用的几种消息模型

第一种模型（HelloWorld）

P代表生产者用来生产消息，发送给消费者C，中间的共色部分代表消息队列，用来缓存消息。

首先导入依赖

<dependency>

    <groupId>com.rabbitmq</groupId>

    <artifactId>amqp-client</artifactId>

    <version>5.7.2</version>

</dependency>

工具类 RabbitmqUtil.java

public class RabbitmqUtil {

    private static ConnectionFactory connectionFactory = new ConnectionFactory();

    /**

     * 配置连接端口，ip,用户名和密码

     */

    static {

        //绑定RabbitMQ主机地址

        connectionFactory.setHost("192.168.1.6");

        //绑定端口

        connectionFactory.setPort(5672);

        //输入用户名密码

        connectionFactory.setUsername("rabbit");

        connectionFactory.setPassword("rabbit");

    }

    /**

     * 获取连接对象

     * @return

     */

    public static Connection getConnection() {

        try {

            return connectionFactory.newConnection();

        } catch (Exception e) {

            e.printStackTrace();

        }

        return null;

    }

    /**

     * 关闭资源

     * @param channel

     * @param connection

     */

    public static void closeConnAndChannel(Channel channel,Connection connection){

        try {

            if (channel != null) {

                channel.close();

            }

            if (connection != null) {

                connection.close();

            }

        }catch (Exception e){

            e.printStackTrace();

        }

    }

}

生产者代码
Provider.java

public class Provider {

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

        // 获取连接对象

        Connection connection = RabbitmqUtil.getConnection();

        // 创建通道

        Channel channel = connection.createChannel();

        //queueDeclare第一个参数是队列名称，第二个是是否持久化，如果是true，队列持久化，但是队列内容的持久化需要在basicProperties设置

        //第三个是是否独占队列一般是false不独占，第四个是消费完成后是否自动删除 true代表删除，false代表不删除

        //第五个参数是额外附加参数

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

        //basicPublish 第一个参数代表交换机名称，第二是队列名称，第三个是额外的队列配置 第四个参数就是发送的消息

        channel.basicPublish("","hello", MessageProperties.PERSISTENT_TEXT_PLAIN,"hello,world".getBytes());

        //关闭管道和连接

        RabbitmqUtil.closeConnAndChannel(channel,connection);

    }

}

Consumer.java

public class Consumer {

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

        // 通过工具类获取连接对象

        Connection connection = RabbitmqUtil.getConnection();

        // 创建通道

        Channel channel = connection.createChannel();

        //  '参数1':用来声明通道对应的队列 hello

        //  '参数2':用来指定是否持久化队列 true

        //  '参数3':用来指定是否独占队列 false 一般都是不独站队列 让多个连接可以共同向一个队列生产消费消息

        //  '参数4':用来指定是否自动删除队列 false

        //  '参数5':对队列的额外配置 是一个Map类型

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

        //参数一代表队列名，参数二是否开启自动确认机制，参数三，消费时的回调接口

        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));

            }

        });

    }

}

第二种模型（Work queue）

C1和C2都是消费者，P代表生产者，中间红色的部分是消息队列。

Work queue被称为任务队列。当消息处理比较耗时时，生产的速度大于消费的速度，长此以往，消息会在消息队列中越来愈多，无法及时处理。此时可以使用work模型，让多了消费者绑定到一个队列中，共同消费队列中的消息。

消息提供者

public class Provider {

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

        Connection connection = RabbitmqUtil.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 queue").getBytes());

        }

        //关闭资源

        RabbitmqUtil.closeConnAndChannel(channel,connection);

    }

}

消费者1

public class Consumer1 {

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

        // 通过工具类获取连接对象

        Connection connection = RabbitmqUtil.getConnection();

        // 创建通道

        Channel channel = connection.createChannel();

        //每次只确认一条消息

        channel.basicQos(1);

        //  '参数1':用来声明通道对应的队列 hello

        //  '参数2':用来指定是否持久化队列 true

        //  '参数3':用来指定是否独占队列 false 一般都是不独站队列 让多个连接可以共同向一个队列生产消费消息

        //  '参数4':用来指定是否自动删除队列 false

        //  '参数5':对队列的额外配置 是一个Map类型

        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));

                //进行手动确认

                channel.basicAck(envelope.getDeliveryTag(),false);

            }

        });

    }

}

消费者2

public class Consumer2 {

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

        // 通过工具类获取连接对象

        Connection connection = RabbitmqUtil.getConnection();

        // 创建通道

        Channel channel = connection.createChannel();

       //一次只接受1条消息进行 确认

        channel.basicQos(1);

        //  '参数1':用来声明通道对应的队列 hello

        //  '参数2':用来指定是否持久化队列 true

        //  '参数3':用来指定是否独占队列 false 一般都是不独站队列 让多个连接可以共同向一个队列生产消费消息

        //  '参数4':用来指定是否自动删除队列 false

        //  '参数5':对队列的额外配置 是一个Map类型

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

        //参数一代表队列名，

        // 参数二是否开启自动确认机，

        // 在开启自动确认消息机制时，RabbitMQ会认为只要消费者从队列中拿走消息就认为已经消费完成，就会将队列中的消息标记为已消费实际消费过程中有可能出现

        //开启自动确认机制有时会造成消息丢失，如果一个消费者在执行过程中宕机了那他未完成的消息也会丢失,我们想让宕机后未消费的消息转移到正常运行的消费者上进行消费，

        // 所以需要关闭自动确认机制（设置成false），进行手动确认

        // 参数三，消费时的回调接口

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

            @lombok.SneakyThrows

            @Override

            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

               // 消费者2消费的速度小于消费者1

               Thread.sleep(2000);

                System.out.println("消费者 2 接收的消息是："+new String(body));

                //进行手动确认消息 false代表不开启多次确认

                channel.basicAck(envelope.getDeliveryTag(),false);

            }

        });

    }

}

上面的代码是进行手动确认消息，默认是自动确认消息，如果是默认的自动消息确认机制，那么消息提供者会将消息平均发送给消费者1和消费者2。也就是说消费者收到的消息数量是相同的，有序的，这种消费发送方式也被叫做循环。

但是我们通常会根据消费者处理的效率不同，让消费快的消费者处理多点消息；消费效率低的消费者处理少量的消息。

第三种模型(Fanout)

扇出(fanout)又称广播
广播模式下，消息发送流程

可以有多个消费者
每个消费者都有自己的队列
每个队列都要绑定到Exchange（交换机）
生产者发送消息，只能发送给交换机，交换机决定要发送到哪个队列，生产者无法决定
交换机把消息发送给绑定过的所有队列
队列的消费者都能够拿到消息，实现一条消息被多个消费者消费
代码实现

消息生产者

public class Provider {

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

        //获取连接对象

        Connection connection = RabbitmqUtil.getConnection();

        //获取管道

        Channel channel = connection.createChannel();

        //将管道绑定交换机

        channel.exchangeDeclare("logs","fanout");

        //发送消息内容

        channel.basicPublish("logs","",null,"fanout 广播模式".getBytes());

        //关闭连接

        RabbitmqUtil.closeConnAndChannel(channel,connection);

    }

}

消费者

public class Consumer1 {

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

        Connection connection = RabbitmqUtil.getConnection();

        Channel channel = connection.createChannel();

        //绑定交换机

        channel.exchangeDeclare("logs","fanout");

        //临时队列

        String queue = channel.queueDeclare().getQueue();

        //绑定交换机和队列

        channel.queueBind(queue,"logs","");

        //消费消息

        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： "+new String(body));

            }

        });

    }

}

其中一个生产者多个消费者进行消费，广播的效果就是可以将同一条消息发送给所有消费者绑定的队列

第四种模式（Routing）

Routing订阅模式-Direct（直连）
在Fanout模式中，一条消息，会被所有订阅的队列都消费但是在某些场景下，我们希望不同的消息被不同的队列消费。这时候就要用到Direct类型的Exchange。
在Direct下：
1、队列与交换机的绑定不能是任意绑定，而是要指定一个RoutingKey（路由Key）
2、消息的发送方在向交换机发送消息时，需要指定RoutingKey。
3、Exchange不在吧消息交给每一个绑定的队列，而是根据消息的Routing Key进行判断，只有队列的RoutingKey与消息的Routing Key完全一致才会接收到消息。

消息生产者

public class Provider {

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

        // 获取连接对象

        Connection connection = RabbitmqUtil.getConnection();

        // 创建通信管道

        Channel channel = connection.createChannel();

        // 交换器名称

        String exechange = "logs_direct";

        channel.exchangeDeclare(exechange,"direct");

        // 指定路由key

        String router_key = "error";

        // 向交换器发送消息

        channel.basicPublish(exechange,router_key,

                null,("这是基于direct模型发布的 route_key="+router_key+" 发送的消息").getBytes());

        RabbitmqUtil.closeConnAndChannel(channel,connection);

    }

}

消费者1

public class Consumer1 {

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

        Connection connection = RabbitmqUtil.getConnection();

        Channel channel = connection.createChannel();

        String exechange = "logs_direct";

        //创建交换机类型是direct

        channel.exchangeDeclare(exechange,"direct");

        //创建临时队列

        String queue = channel.queueDeclare().getQueue();

        //绑定交换机，指明路由id

        channel.queueBind(queue,exechange,"info");

        channel.queueBind(queue,exechange,"warning");

        channel.queueBind(queue,exechange,"error");

        //消费消息

        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： "+new String(body));

            }

        });

    }

}

消费者2

public class Consumer2 {

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

        Connection connection = RabbitmqUtil.getConnection();

        Channel channel = connection.createChannel();

        String exechange = "logs_direct";

        //创建交换机类型是direct

        channel.exchangeDeclare(exechange,"direct");

        //创建临时队列

        String queue = channel.queueDeclare().getQueue();

        //绑定交换机，指明路由id

        channel.queueBind(queue,exechange,"error");

        //消费消息

        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： "+new String(body));

            }

        });

    }

}

第五种模型（Topic）

Topic模型的Exchange与Direct相比，是可以根据RoutingKey的不同把消息路由到不同的队列。只不过Topic类型的Exchange可以让队列在绑定路由key的时候使用通配符，这种模型的路由key一般都是由一个或多个单词组成，多个单词之间以“.”分割。

# 通配符

* 表示可以匹配一个单词

\# 表示可以匹配一个或多个单词

例如：user.\#  可以代表user.login或者user.logout ，user.register.check等

user.* 表示user.login；user.lgout但是不能表示user.register.check

生产者

public class Provider {

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

        // 获取连接对象

        Connection connection = RabbitmqUtil.getConnection();

        // 创建通信管道

        Channel channel = connection.createChannel();

        // 交换器名称

        String exechange = "topics";

        channel.exchangeDeclare(exechange,"topic");

        // 指定路由key

        String router_key = "response.error";

        // 向交换器发送消息

        channel.basicPublish(exechange,router_key,

                null,("这是基于topic模型发布的 route_key="+router_key+" 发送的消息").getBytes());

        RabbitmqUtil.closeConnAndChannel(channel,connection);

    }

}

消费者1

public class Consumer1 {

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

        Connection connection = RabbitmqUtil.getConnection();

        Channel channel = connection.createChannel();

        String exechange = "topics";

        //创建交换机类型是direct

        channel.exchangeDeclare(exechange,"topic");

        //创建临时队列

        String queue = channel.queueDeclare().getQueue();

        //绑定交换机，指明路由id

        channel.queueBind(queue,exechange,"response.*");

        //消费消息

        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： "+new String(body));

            }

        });

    }

}

消费者2



public class Consumer2 {

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

        Connection connection = RabbitmqUtil.getConnection();

        Channel channel = connection.createChannel();

        String exechange = "topics";

        //创建交换机类型是direct

        channel.exchangeDeclare(exechange,"topic");

        //创建临时队列

        String queue = channel.queueDeclare().getQueue();

        //绑定交换机，指明路由id

        channel.queueBind(queue,exechange,"response.#");

        //消费消息

        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： "+new String(body));

            }

        });

    }

}

运行结果

路由key改为response.error.code

巴特西

RabbitMQ常用的几种消息模型

第一种模型（HelloWorld）

第二种模型（Work queue）

第三种模型(Fanout)

第四种模式（Routing）

第五种模型（Topic）

最新文章

热门文章