RabbitMQ快速入门及六大模式

2023-05-16

目录

核心组件

运行原理

实现步骤

构建Gradle项目

入门案例(简单模式)

生产者（代码）

消费者（代码）

绑定交换机和队列

发布/订阅模式（Publish/Subscribe）

生产者

消费者

路由模式(routing)

生产者

消费者

主题模式(topics)

生产者

消费者

工作模式(Work queue)

轮询模式

公平分配模式

核心组件

- Server:又称Broker接受客户端的连接，实现AMQP实体服务。安装rabbitmq-server Connection:连接，应用程序与Broker的网络连接TCP/IP/三次握手和四次挥手

- Channel:网络信道，几乎所有的操作都在Channel中进行，Channel是进行消息读写的通道，客户端可以建立对各 Channel，每个Channel代表一个会话任务。

- Message:消息:服务与应用程序之间传送的数据，由Properties和body组成，Properties可是对消息进行修饰，比如消息的优先级，延迟等高级特性，Body则就是消息体的内容。

- VirtualHost虚拟地址，用于进行逻辑隔离，最上层的消息路由，一个虚拟主机理由可以有若干个Exhange和 Queueu，同一个虚拟王机里面不能有相同名字的Exchange

- Exchange:交换机，接受消息，根据路由键发送消息到绑定的队列。(==不具备消息存储的能力==)

- Bindings:Exchange和Queue之间的虚拟连接，binding中可以保护多个routing key.

- Routingkey:是一个路由规则，虚拟机可以用它来确定如何路由一个特定消息。

- Queue:队列:也成为MessageQueue,消息队列，保存消息并将它们转发给消费者。

运行原理

实现步骤

1、JDK11
2、构建一个gradle项目
3、导入RabbitMQ的相关依赖
4、启动rabbitmq-server服务
5、定义生产者
6、定义消费者
7、查看消息在rabbitmq-server服务中的过程

构建Gradle项目

Maven 同样引入相应的依赖

implementation group: 'com.rabbitmq', name: 'amqp-client', version: '5.10.0'

入门案例(简单模式)

生产者（代码）

package com.any.rabbitmq.simple;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Producer {

    public static void main(String[] args) {
        //所有中间件的技术都是基于tcp/ip 协议的基础之上构建新型的协议规范，只不过rabbitmq 遵循的amqp
        // ip port
        // 1、创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setUsername("admin");
        factory.setPassword("admin");
        factory.setVirtualHost("/");    // 虚拟访问节点
        Connection connection = null;
        Channel channel = null;
        try {
            // 2、创建连接Connection
            connection = factory.newConnection("生产者");
            // 3、通过连接获取通道Channel
            channel = connection.createChannel();
            String queueName = "queue1";
            /**
             * @params1 队列的名称
             * @params2 是否要持久化durable=false 所有的持久化是否存盘，如果false 不持久化 ture 持久化
             *          非持久化会存盘吗？ 会存盘，随着服务器的存盘会丢失
             * @params3 排他性，是否是独占队列
             * @params4 是否自动删除，随着最后一个消费者消费消息完毕，是否把队列自动删除
             * @params5 携带一些附加参数
             */
            // 4、通过通道交换机，队列，绑定关系，路由key,发送消息，和接收消息
            channel.queueDeclare(queueName,false,false,true,null);
            // 5、准备消息
            String message = "hello,anyboot";
            // 6、发送消息给 queue
            channel.basicPublish("",queueName,null,message.getBytes());
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if(channel !=null && channel.isOpen()){
                try {
                    // 7、关闭连接
                    channel.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            if(connection!=null && connection.isOpen()){
                try {
                    // 8、关闭通道
                    connection.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

消费者（代码）

package com.any.rabbitmq.simple;

import com.rabbitmq.client.*;
import java.io.IOException;

/**
 * 消费者
 *
 * @author 15821
 * @date 20:20 2022/1/6
 */
public class Consumer {
    public static void main(String[] args) {
        //所有中间件的技术都是基于tcp/ip 协议的基础之上构建新型的协议规范，只不过rabbitmq 遵循的amqp
        // ip port
        // 1、创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("101.35.118.177");
        factory.setPort(5672);
        factory.setUsername("admin");
        factory.setPassword("admin");
        factory.setVirtualHost("/");    // 虚拟访问节点
        Connection connection = null;
        Channel channel = null;
        try {
            // 2、创建连接Connection
            connection = factory.newConnection("生产者");
            // 3、通过连接获取通道Channel
            channel = connection.createChannel();
            String queueName = "queue1";
            channel.basicConsume(queueName, true, new DeliverCallback() {
                @Override
                public void handle(String consumerTag, Delivery message) throws IOException {
                    System.out.println("收到的消息：" + new String(message.getBody(), "UTF-8"));
                }
            }, new CancelCallback() {
                @Override
                public void handle(String consumerTag) throws IOException {
                    System.out.println("消息接受失败");
                }
            });
            System.in.read();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if(channel !=null && channel.isOpen()){
                try {
                    // 7、关闭连接
                    channel.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            if(connection!=null && connection.isOpen()){
                try {
                    // 8、关闭通道
                    connection.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

绑定交换机和队列

该例子以生产者为例

package com.any.rabbitmq.all;

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

import java.io.IOException;

/**
 * 路由模式
 * 交换机声明，队列声明， 绑定(binding)
 * 生产者
 *
 */
public class Producer {

    public static void main(String[] args) {
        //所有中间件的技术都是基于tcp/ip 协议的基础之上构建新型的协议规范，只不过rabbitmq 遵循的amqp
        // ip port

        // 1、创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("101.35.118.177");
        factory.setPort(5672);
        factory.setUsername("admin");
        factory.setPassword("admin");
        factory.setVirtualHost("/");    // 虚拟访问节点

        Connection connection = null;
        Channel channel = null;
        try {
            // 2、创建连接Connection
            connection = factory.newConnection("生产者");
            // 3、通过连接获取信道Channel
            channel = connection.createChannel();

            // 定义交换机名称
            String exchangeName = "direct-message-exchange";

            // 定义交换机类型
            String exchangeType = "direct";

            // 定义 路由key
            String routingKey = "message";

            /**
             * 4、声明交换机
             * @param1 exchange 交换机名称
             * @param2 type 交换机类型
             * @param3 durable 重启是否删除
             */
            channel.exchangeDeclare(exchangeName,exchangeType,true);
            /**
             * 5、声明队列
             * @params1 队列的名称
             * @params2 是否要持久化durable=false 所有的持久化是否存盘，如果false 不持久化 ture 持久化
             *          非持久化会存盘吗？ 会存盘，随着服务器的存盘会丢失
             * @params3 排他性，是否是独占队列
             * @params4 是否自动删除，随着最后一个消费者消费消息完毕，是否把队列自动删除
             * @params5 携带一些附加参数
             */

            channel.queueDeclare("queue5",true,false,false,null);
            channel.queueDeclare("queue6",true,false,false,null);
            channel.queueDeclare("queue7",true,false,false,null);

            /**
             * 6、绑定
             * @param1 queue 队列名称
             * @param2 exchange 交换机名称
             * @param3 routingKey 路由Key
             */
            channel.queueBind("queue5",exchangeName,routingKey);
            channel.queueBind("queue6",exchangeName,routingKey);
            channel.queueBind("queue7",exchangeName,"email");

            // 5、准备消息
            String message = "hello,message-exchange";
            // 6、发送消息给 queue
            /**
             * @params1 exchange 交换机
             * @params2 队列名称
             * @params3 props 消息状态控制
             * @params4 body 消息内容
             */
            channel.basicPublish(exchangeName,routingKey,null,message.getBytes());
        } catch (Exception e) {
            e.printStackTrace();
        } finally {

            if(channel !=null && channel.isOpen()){
                try {
                    // 7、关闭连接
                    channel.close();

                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            if(connection!=null && connection.isOpen()){
                try {
                    // 8、关闭通道
                    connection.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }

        }
    }

}

发布/订阅模式（Publish/Subscribe）

生产者

package com.any.rabbitmq.routing.fanout;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;

/**
 * 发布/订阅模式
 * fanout 模式
 * 生产者
 * @author 15821
 * @date 20:20 2022/1/6
 */
public class Producer {
    // 发布订阅模式
    public static void main(String[] args) {
        // 1、创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("101.35.118.177");
        factory.setPort(5672);
        factory.setUsername("admin");
        factory.setPassword("admin");
        factory.setVirtualHost("/");    // 虚拟访问节点
        Connection connection = null;
        Channel channel = null;
        try {
            // 2、创建连接Connection
            connection = factory.newConnection("生产者");
            // 3、通过连接获取通道Channel
            channel = connection.createChannel();
            // 定义 队列名称
            String queueName = "queue1";
            // 定义交换机名称
            String exchangeName = "fanout-exchange";
            // 定义交换机类型
            String exchangeType = "fanout";
            // 定义 路由key
            String routingKey = "";
            /**
             * @params1 队列的名称
             * @params2 是否要持久化durable=false 所有的持久化是否存盘，如果false 不持久化 ture 持久化
             *          非持久化会存盘吗？ 会存盘，随着服务器的存盘会丢失
             * @params3 排他性，是否是独占队列
             * @params4 是否自动删除，随着最后一个消费者消费消息完毕，是否把队列自动删除
             * @params5 携带一些附加参数
             */
            // 4、通过通道交换机，队列，绑定关系，路由key,发送消息，和接收消息
            //channel.queueDeclare(queueName,true,false,true,null);
            // 5、准备消息
            String message = "hello,anyboot";
            // 6、发送消息给 queue
            /**
             * @params1 exchange 交换机
             * @params2 队列名称
             * @params3 props 消息状态控制
             * @params4 body 消息内容
             */
            channel.basicPublish(exchangeName,routingKey,null,message.getBytes());
        } catch (Exception e) {
            e.printStackTrace();
        } finally {

            if(channel !=null && channel.isOpen()){
                try {
                    // 7、关闭连接
                    channel.close();

                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            if(connection!=null && connection.isOpen()){
                try {
                    // 8、关闭通道
                    connection.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }

        }
    }

}

消费者

package com.any.rabbitmq.routing.fanout;

import com.rabbitmq.client.*;

import java.io.IOException;

/**
 * 发布/订阅模式Publish/Subscribe
 * fanout 模式
 * 消费者
 *
 * @author 15821
 * @date 20:20 2022/1/6
 */
public class Consumer {

    public static Runnable runnable = new Runnable() {
        @Override
        public void run() {
            //所有中间件的技术都是基于tcp/ip 协议的基础之上构建新型的协议规范，只不过rabbitmq 遵循的amqp
            // ip port
            // 1、创建连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("101.35.118.177");
            factory.setPort(5672);
            factory.setUsername("admin");
            factory.setPassword("admin");
            factory.setVirtualHost("/");    // 虚拟访问节点

            final String queueName = Thread.currentThread().getName();
            Connection connection = null;
            Channel channel = null;
            try {
                // 2、创建连接Connection
                connection = factory.newConnection("生产者");
                // 3、通过连接获取通道Channel
                channel = connection.createChannel();
                channel.basicConsume( queueName , true, new DeliverCallback() {
                    // 成功处理
                    @Override
                    public void handle(String consumerTag, Delivery message) throws IOException {
                        // 获取队列中消息的数量
                        System.out.println(message.getEnvelope().getDeliveryTag());
                        System.out.println(queueName + "-收到的消息：" + new String(message.getBody(), "UTF-8"));
                    }
                }, new CancelCallback() {
                    // 失败处理
                    @Override
                    public void handle(String consumerTag) throws IOException {
                        System.out.println("消息接受失败");
                    }
                });

                System.out.println(queueName+"-开始接受消息");
                System.in.read();
            } catch (Exception e) {
                e.printStackTrace();
            } finally {

                if(channel !=null && channel.isOpen()){
                    try {
                        // 7、关闭连接
                        channel.close();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
                if(connection!=null && connection.isOpen()){
                    try {
                        // 8、关闭通道
                        connection.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    };

    public static void main(String[] args) {
        new Thread(runnable,"queue1").start();
        new Thread(runnable,"queue2").start();
        new Thread(runnable,"queue3").start();
    }
}

路由模式(routing)

生产者

package com.any.rabbitmq.routing.direct;

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

import java.io.IOException;

/**
 * 路由模式
 * routing 模式
 * 生产者
 *
 * @author 15821
 * @date 20:20 2022/1/6
 */
public class Producer {

    public static void main(String[] args) {
        //所有中间件的技术都是基于tcp/ip 协议的基础之上构建新型的协议规范，只不过rabbitmq 遵循的amqp
        // ip port

        // 1、创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("101.35.118.177");
        factory.setPort(5672);
        factory.setUsername("admin");
        factory.setPassword("admin");
        factory.setVirtualHost("/");    // 虚拟访问节点

        Connection connection = null;
        Channel channel = null;
        try {
            // 2、创建连接Connection
            connection = factory.newConnection("生产者");
            // 3、通过连接获取通道Channel
            channel = connection.createChannel();



            // 定义 队列名称
            String queueName = "queue1";


            // 定义交换机名称
            String exchangeName = "direct-exchange";

            // 定义交换机类型
            String exchangeType = "direct";

            // 定义 路由key
            String routingKey = "email";
            /**
             * @params1 队列的名称
             * @params2 是否要持久化durable=false 所有的持久化是否存盘，如果false 不持久化 ture 持久化
             *          非持久化会存盘吗？ 会存盘，随着服务器的存盘会丢失
             * @params3 排他性，是否是独占队列
             * @params4 是否自动删除，随着最后一个消费者消费消息完毕，是否把队列自动删除
             * @params5 携带一些附加参数
             */
            // 4、通过通道交换机，队列，绑定关系，路由key,发送消息，和接收消息
            //channel.queueDeclare(queueName,true,false,true,null);
            // 5、准备消息
            String message = "hello,anyboot";
            // 6、发送消息给 queue
            /**
             * @params1 exchange 交换机
             * @params2 队列名称
             * @params3 props 消息状态控制
             * @params4 body 消息内容
             */
            channel.basicPublish(exchangeName,routingKey,null,message.getBytes());
        } catch (Exception e) {
            e.printStackTrace();
        } finally {

            if(channel !=null && channel.isOpen()){
                try {
                    // 7、关闭连接
                    channel.close();

                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            if(connection!=null && connection.isOpen()){
                try {
                    // 8、关闭通道
                    connection.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }

        }


    }

}

消费者

package com.any.rabbitmq.routing.direct;

import com.rabbitmq.client.*;

import java.io.IOException;

/**
 * 路由模式
 * routing 模式
 * 费者
 *
 * @author 15821
 * @date 20:20 2022/1/6
 */
public class Consumer {

    public static Runnable runnable = new Runnable() {
        @Override
        public void run() {
            //所有中间件的技术都是基于tcp/ip 协议的基础之上构建新型的协议规范，只不过rabbitmq 遵循的amqp
            // ip port
            // 1、创建连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("101.35.118.177");
            factory.setPort(5672);
            factory.setUsername("admin");
            factory.setPassword("admin");
            factory.setVirtualHost("/");    // 虚拟访问节点

            final String queueName = Thread.currentThread().getName();
            Connection connection = null;
            Channel channel = null;
            try {
                // 2、创建连接Connection
                connection = factory.newConnection("生产者");
                // 3、通过连接获取通道Channel
                channel = connection.createChannel();
                channel.basicConsume( queueName , true, new DeliverCallback() {
                    // 成功处理
                    @Override
                    public void handle(String consumerTag, Delivery message) throws IOException {
                        // 获取队列中消息的数量
                        System.out.println(message.getEnvelope().getDeliveryTag());
                        System.out.println(queueName + "-收到的消息：" + new String(message.getBody(), "UTF-8"));
                    }
                }, new CancelCallback() {
                    // 失败处理
                    @Override
                    public void handle(String consumerTag) throws IOException {
                        System.out.println("消息接受失败");
                    }
                });

                System.out.println(queueName+"-开始接受消息");
                System.in.read();
            } catch (Exception e) {
                e.printStackTrace();
            } finally {

                if(channel !=null && channel.isOpen()){
                    try {
                        // 7、关闭连接
                        channel.close();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
                if(connection!=null && connection.isOpen()){
                    try {
                        // 8、关闭通道
                        connection.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    };

    public static void main(String[] args) {
        new Thread(runnable,"queue1").start();
        new Thread(runnable,"queue2").start();
        new Thread(runnable,"queue3").start();
    }
}

主题模式(topics)

生产者

package com.any.rabbitmq.routing.topics;

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

import java.io.IOException;

/**
 * 主题模式
 * topics 模式
 * 生产者
 *
 * @author 15821
 * @date 20:20 2022/1/6
 */
public class Producer {

    public static void main(String[] args) {

        // 1、创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("101.35.118.177");
        factory.setPort(5672);
        factory.setUsername("admin");
        factory.setPassword("admin");
        factory.setVirtualHost("/");    // 虚拟访问节点

        Connection connection = null;
        Channel channel = null;
        try {
            // 2、创建连接Connection
            connection = factory.newConnection("生产者");
            // 3、通过连接获取通道Channel
            channel = connection.createChannel();

            // 定义 队列名称
            String queueName = "queue1";

            // 定义交换机名称
            String exchangeName = "topic-exchange";

            // 定义交换机类型
            String exchangeType = "topic";
            // 定义 路由key
            String routingKey = "com.course.order";
            /**
             * @params1 队列的名称
             * @params2 是否要持久化durable=false 所有的持久化是否存盘，如果false 不持久化 ture 持久化
             *          非持久化会存盘吗？ 会存盘，随着服务器的存盘会丢失
             * @params3 排他性，是否是独占队列
             * @params4 是否自动删除，随着最后一个消费者消费消息完毕，是否把队列自动删除
             * @params5 携带一些附加参数
             */
            // 4、通过通道交换机，队列，绑定关系，路由key,发送消息，和接收消息
            //channel.queueDeclare(queueName,true,false,true,null);
            // 5、准备消息
            String message = "hello,anyboot";
            // 6、发送消息给 queue
            /**
             * @params1 exchange 交换机
             * @params2 队列名称
             * @params3 props 消息状态控制
             * @params4 body 消息内容
             */
            channel.basicPublish(exchangeName,routingKey,null,message.getBytes());
        } catch (Exception e) {
            e.printStackTrace();
        } finally {

            if(channel !=null && channel.isOpen()){
                try {
                    // 7、关闭连接
                    channel.close();

                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            if(connection!=null && connection.isOpen()){
                try {
                    // 8、关闭通道
                    connection.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }

        }


    }

}

消费者

package com.any.rabbitmq.routing.topics;

import com.rabbitmq.client.*;

import java.io.IOException;

/**
 * 主题模式
 * topics 模式
 * 消费者
 *
 * @author 15821
 * @date 20:20 2022/1/6
 */
public class Consumer {

    public static Runnable runnable = new Runnable() {
        @Override
        public void run() {
            //所有中间件的技术都是基于tcp/ip 协议的基础之上构建新型的协议规范，只不过rabbitmq 遵循的amqp
            // ip port
            // 1、创建连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("101.35.118.177");
            factory.setPort(5672);
            factory.setUsername("admin");
            factory.setPassword("admin");
            factory.setVirtualHost("/");    // 虚拟访问节点

            final String queueName = Thread.currentThread().getName();
            Connection connection = null;
            Channel channel = null;
            try {
                // 2、创建连接Connection
                connection = factory.newConnection("生产者");
                // 3、通过连接获取通道Channel
                channel = connection.createChannel();
                channel.basicConsume( queueName , true, new DeliverCallback() {
                    // 成功处理
                    @Override
                    public void handle(String consumerTag, Delivery message) throws IOException {
                        // 获取队列中消息的数量
                        System.out.println(message.getEnvelope().getDeliveryTag());
                        System.out.println(queueName + "-收到的消息：" + new String(message.getBody(), "UTF-8"));
                    }
                }, new CancelCallback() {
                    // 失败处理
                    @Override
                    public void handle(String consumerTag) throws IOException {
                        System.out.println("消息接受失败");
                    }
                });

                System.out.println(queueName+"-开始接受消息");
                System.in.read();
            } catch (Exception e) {
                e.printStackTrace();
            } finally {

                if(channel !=null && channel.isOpen()){
                    try {
                        // 7、关闭连接
                        channel.close();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
                if(connection!=null && connection.isOpen()){
                    try {
                        // 8、关闭通道
                        connection.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    };

    public static void main(String[] args) {
        new Thread(runnable,"queue1").start();
        new Thread(runnable,"queue2").start();
        new Thread(runnable,"queue3").start();
    }
}

工作模式(Work queue)

轮询模式

生产者

package com.any.rabbitmq.work.polling;

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

import java.io.IOException;

/**
 * work模式 轮询分发
 * 生产者
 *
 * @author 15821
 * @date 20:20 2022/1/6
 */
public class Producer {

    public static void main(String[] args) {
        // 1、创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("101.35.118.177");
        factory.setPort(5672);
        factory.setUsername("admin");
        factory.setPassword("admin");
        factory.setVirtualHost("/");    // 虚拟访问节点

        Connection connection = null;
        Channel channel = null;
        try {
            // 2、创建连接Connection
            connection = factory.newConnection("生产者");
            // 3、通过连接获取通道Channel
            channel = connection.createChannel();

            // 定义 队列名称
            String queueName = "queue8";

            // 4、发送消息给 queue
            for (int i = 0; i < 20; i++) {
                String message = "hello,work-rabbitmq - " + i;
                /**
                 * @params1 exchange 交换机
                 * @params2 队列名称
                 * @params3 props 消息状态控制
                 * @params4 body 消息内容
                 */
                channel.basicPublish("",queueName,null,message.getBytes());
            }

        } catch (Exception e) {
            e.printStackTrace();
        } finally {

            if(channel !=null && channel.isOpen()){
                try {
                    // 7、关闭连接
                    channel.close();

                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            if(connection!=null && connection.isOpen()){
                try {
                    // 8、关闭通道
                    connection.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }

        }


    }

}

work1

package com.any.rabbitmq.work.polling;

import com.rabbitmq.client.*;

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

/**
 * work 模式  轮询分发
 * 消费者
 *
 * @author 15821
 * @date 20:20 2022/1/6
 */
public class Work1 {

    public static Runnable runnable = new Runnable() {
        @Override
        public void run() {
            // 1、创建连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("101.35.118.177");
            factory.setPort(5672);
            factory.setUsername("admin");
            factory.setPassword("admin");
            factory.setVirtualHost("/");    // 虚拟访问节点

            final String queueName = Thread.currentThread().getName();
            Connection connection = null;
            Channel channel = null;
            try {
                // 2、创建连接Connection
                connection = factory.newConnection("生产者");
                // 3、通过连接获取通道Channel
                channel = connection.createChannel();
                channel.basicConsume( queueName , true, new DeliverCallback() {
                    // 成功处理
                    @Override
                    public void handle(String consumerTag, Delivery message) throws IOException {
                        // 获取队列中消息的数量
                        System.out.println(message.getEnvelope().getDeliveryTag());
                        System.out.println(queueName + "-收到的消息：" + new String(message.getBody(), "UTF-8"));
                        try {
                            TimeUnit.SECONDS.sleep(1);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }, new CancelCallback() {
                    // 失败处理
                    @Override
                    public void handle(String consumerTag) throws IOException {
                        System.out.println("消息接受失败");
                    }
                });

                System.out.println(queueName+"-开始接受消息");
                System.in.read();
            } catch (Exception e) {
                e.printStackTrace();
            } finally {

                if(channel !=null && channel.isOpen()){
                    try {
                        // 7、关闭连接
                        channel.close();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
                if(connection!=null && connection.isOpen()){
                    try {
                        // 8、关闭通道
                        connection.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    };

    public static void main(String[] args) {
        new Thread(runnable,"queue8").start();
    }
}

work2

package com.any.rabbitmq.work.polling;

import com.rabbitmq.client.*;

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

/**
 * work 模式  轮询分发
 * 消费者
 *
 * @author 15821
 * @date 20:20 2022/1/6
 */
public class Work2 {

    public static Runnable runnable = new Runnable() {
        @Override
        public void run() {
            //所有中间件的技术都是基于tcp/ip 协议的基础之上构建新型的协议规范，只不过rabbitmq 遵循的amqp
            // ip port
            // 1、创建连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("101.35.118.177");
            factory.setPort(5672);
            factory.setUsername("admin");
            factory.setPassword("admin");
            factory.setVirtualHost("/");    // 虚拟访问节点

            final String queueName = Thread.currentThread().getName();
            Connection connection = null;
            Channel channel = null;
            try {
                // 2、创建连接Connection
                connection = factory.newConnection("生产者");
                // 3、通过连接获取通道Channel
                channel = connection.createChannel();
                //channel.
                channel.basicConsume( queueName , true, new DeliverCallback() {
                    // 成功处理
                    @Override
                    public void handle(String consumerTag, Delivery message) throws IOException {
                        // 获取队列中消息的数量
                        System.out.println(message.getEnvelope().getDeliveryTag());
                        System.out.println(queueName + "-收到的消息：" + new String(message.getBody(), "UTF-8"));
                        try {
                            TimeUnit.SECONDS.sleep(2);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }, new CancelCallback() {
                    // 失败处理
                    @Override
                    public void handle(String consumerTag) throws IOException {
                        System.out.println("消息接受失败");
                    }
                });

                System.out.println(queueName+"-开始接受消息");
                System.in.read();
            } catch (Exception e) {
                e.printStackTrace();
            } finally {

                if(channel !=null && channel.isOpen()){
                    try {
                        // 7、关闭连接
                        channel.close();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
                if(connection!=null && connection.isOpen()){
                    try {
                        // 8、关闭通道
                        connection.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    };

    public static void main(String[] args) {
        new Thread(runnable,"queue8").start();
    }
}

公平分配模式

package com.any.rabbitmq.work.fair;

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

import java.io.IOException;

/**
 * work模式 公平分配
 * 生产者
 *
 * @author 15821
 * @date 20:20 2022/1/6
 */
public class Producer {

    public static void main(String[] args) {
        //所有中间件的技术都是基于tcp/ip 协议的基础之上构建新型的协议规范，只不过rabbitmq 遵循的amqp
        // ip port

        // 1、创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("101.35.118.177");
        factory.setPort(5672);
        factory.setUsername("admin");
        factory.setPassword("admin");
        factory.setVirtualHost("/");    // 虚拟访问节点

        Connection connection = null;
        Channel channel = null;
        try {
            // 2、创建连接Connection
            connection = factory.newConnection("生产者");
            // 3、通过连接获取通道Channel
            channel = connection.createChannel();

            // 定义 队列名称
            String queueName = "queue8";

            // 4、发送消息给 queue
            for (int i = 0; i < 20; i++) {
                String message = "hello,work-rabbitmq - " + i;
                /**
                 * @params1 exchange 交换机
                 * @params2 队列名称
                 * @params3 props 消息状态控制
                 * @params4 body 消息内容
                 */
                channel.basicPublish("",queueName,null,message.getBytes());
            }

        } catch (Exception e) {
            e.printStackTrace();
        } finally {

            if(channel !=null && channel.isOpen()){
                try {
                    // 7、关闭连接
                    channel.close();

                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            if(connection!=null && connection.isOpen()){
                try {
                    // 8、关闭通道
                    connection.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }

        }


    }

}

work1

package com.any.rabbitmq.work.fair;

import com.rabbitmq.client.*;
import lombok.SneakyThrows;

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

/**
 * work 模式  公平分配
 * work 模式
 * 消费者
 *
 * @author 15821
 * @date 20:20 2022/1/6
 */
public class Work1 {

    public static Runnable runnable = new Runnable() {
        @Override
        public void run() {
            //所有中间件的技术都是基于tcp/ip 协议的基础之上构建新型的协议规范，只不过rabbitmq 遵循的amqp
            // ip port
            // 1、创建连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("101.35.118.177");
            factory.setPort(5672);
            factory.setUsername("admin");
            factory.setPassword("admin");
            factory.setVirtualHost("/");    // 虚拟访问节点

            final String queueName = Thread.currentThread().getName();
            Connection connection = null;
            Channel channel = null;
            try {
                // 2、创建连接Connection
                connection = factory.newConnection("生产者");
                // 3、通过连接获取通道Channel
                channel = connection.createChannel();
                Channel finalChannel = channel;
                // 确认指标，一次性处理多少条消息
                finalChannel.basicQos(1);

                finalChannel.basicConsume( queueName , false, new DeliverCallback() {
                    // 成功处理
                    @SneakyThrows
                    @Override
                    public void handle(String consumerTag, Delivery message) throws IOException {
                        // 获取队列中消息的数量
                        System.out.println(message.getEnvelope().getDeliveryTag());
                        System.out.println(queueName + "-收到的消息：" + new String(message.getBody(), "UTF-8"));
                        Thread.sleep(1000);
                        // 设置手动应答
                        finalChannel.basicAck(message.getEnvelope().getDeliveryTag(),false);

                    }
                }, new CancelCallback() {
                    // 失败处理
                    @Override
                    public void handle(String consumerTag) throws IOException {
                        System.out.println("消息接受失败");
                    }
                });

                System.out.println(queueName+"-开始接受消息");
                System.in.read();
            } catch (Exception e) {
                e.printStackTrace();
            } finally {

                if(channel !=null && channel.isOpen()){
                    try {
                        // 7、关闭连接
                        channel.close();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
                if(connection!=null && connection.isOpen()){
                    try {
                        // 8、关闭通道
                        connection.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    };

    public static void main(String[] args) {
        new Thread(runnable,"queue8").start();
    }
}

work2

package com.any.rabbitmq.work.fair;

import com.rabbitmq.client.*;
import lombok.SneakyThrows;

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

/**
 * work 模式  公平分配
 * 消费者
 *
 * @author 15821
 * @date 20:20 2022/1/6
 */
public class Work2 {

    public static Runnable runnable = new Runnable() {
        @Override
        public void run() {
            //所有中间件的技术都是基于tcp/ip 协议的基础之上构建新型的协议规范，只不过rabbitmq 遵循的amqp
            // ip port
            // 1、创建连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("101.35.118.177");
            factory.setPort(5672);
            factory.setUsername("admin");
            factory.setPassword("admin");
            factory.setVirtualHost("/");    // 虚拟访问节点

            final String queueName = Thread.currentThread().getName();
            Connection connection = null;
            Channel channel = null;
            try {
                // 2、创建连接Connection
                connection = factory.newConnection("生产者");
                // 3、通过连接获取通道Channel
                channel = connection.createChannel();
                Channel finalChannel = channel;
                finalChannel.basicQos(1);

                finalChannel.basicConsume( queueName , false, new DeliverCallback() {
                    // 成功处理
                    @SneakyThrows
                    @Override
                    public void handle(String consumerTag, Delivery message) throws IOException {
                        // 获取队列中消息的数量
                        System.out.println(message.getEnvelope().getDeliveryTag());
                        System.out.println(queueName + "-收到的消息：" + new String(message.getBody(), "UTF-8"));
                        Thread.sleep(2000);
                        // 设置手动应答
                        finalChannel.basicAck(message.getEnvelope().getDeliveryTag(),false);

                    }
                }, new CancelCallback() {
                    // 失败处理
                    @Override
                    public void handle(String consumerTag) throws IOException {
                        System.out.println("消息接受失败");
                    }
                });

                System.out.println(queueName+"-开始接受消息");
                System.in.read();
            } catch (Exception e) {
                e.printStackTrace();
            } finally {

                if(channel !=null && channel.isOpen()){
                    try {
                        // 7、关闭连接
                        channel.close();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
                if(connection!=null && connection.isOpen()){
                    try {
                        // 8、关闭通道
                        connection.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    };

    public static void main(String[] args) {
        new Thread(runnable,"queue8").start();
    }
}

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)

RabbitMQ快速入门及六大模式的相关文章

HJ71 字符串通配符

描述问题描述 xff1a 在计算机中 xff0c 通配符一种特殊语法 xff0c 广泛应用于文件搜索数据库正则表达式等领域现要求各位实现字符串通配符的算法要求 xff1a 实现如下2个通配符 xff1a xff1a 匹配0个或以上
如何用Java实现判断一个链表是否有环

设置一个快指针和一个慢指针 xff0c 快指针一次走两步 xff0c 慢指针一次走一步如果该链表没有环 xff0c 快指针会先指向NULL xff0c 可据此判断链表没有环结构 xff1b 如果该链表有环 xff0c 则快指针肯定先进环
Arch Linux安装 2023-04-09

除特殊场景外下面所有的 34 34 后都是注释准备 EFI分区不小于200mb iso版本 archlinux 2023 04 01 x86 64 使用分区工具预留出足够空间 gt 20G 标签为未分配安装连接网络 rfkill
Spring Boot配置数据库链接池

配置方法基于当前的1 5 2 RELEASE的Spring Boot 依照官方文档 xff0c 如果增加了如下依赖的配置 xff0c 或者类路径中存在spring boot starter jdbc的jar xff0c 那么已默认启用了数
一文搞懂Java中相对路径与绝对路径

一文搞懂Java中相对路径与绝对路径在java中路径分隔使用正斜杠 xff0c 不推荐使用反斜杠 xff08 因为反斜杠需要转义两个反斜杠表示一个正斜杠 xff09 在windows磁盘中用反斜杠表示路径的分隔在浏览器中用正斜杠来
【向量的叉乘】

一二维向量叉乘公式 xff1a a xff08 x1 xff0c y1 xff09 xff0c b xff08 x2 xff0c y2 xff09 xff0c 则a b 61 xff08 x1y2 x2y1 xff09 二 a b 61
JAVA常用类

Object类 Object类是类层次结构的根 xff0c 每个类都可以将Object作为超类所有类都直接或者间接的继承该类 Object只有无参构造方法 Math类包含执行基本数学运算的方法 Random类伪随机数 java uta
Collections类 [Java]

Collections工具类 Collections是一个操作Collection集合和Map集合的工具类 Collections不仅仅是操作Collection集合还可以操作Map集合 Collection和Collections有什么
我阿里P7了解到的Android面试的一些小内幕！已拿offer

前言这些题目是网友去百度小米乐视美团 58 猎豹 360 新浪搜狐等一线互联网公司面试被问到的题目熟悉本文中列出的知识点会大大增加通过前两轮技术面试的几率欢迎一线公司员工以及网友提交面试题库 xff0c 欢迎留言网上的都是按
7月编程排行榜来啦！这次有何新变化？

每月编程排行榜可能会迟到 xff0c 但永远不缺席 7月的编程排行榜已出 xff0c 接下来一起看看有哪些看点吧 Tiobe编程排行榜前20名 Tiobe编程排行榜Top 10趋势 TIOBE Index编程社区指数是编程语言流行度的一个指
操作系统记录型信号量实现生产者消费者问题（完整代码）

问题描述用信号量模拟生产者消费者问题的过程生产者和消费者两个线程共享同一个缓冲区 xff0c 生产者不断向缓冲区中添加产品 xff0c 消费者从缓冲区中消费产品要保证缓冲区满了之后生产者不能再继续添加产品 xff0c 需要等消费者至
制版经验分享—使用AD18

文章目录前言一封装二走线三注意细节四制版流程五制版细节总结前言在做一些培训题目时 xff0c 由于时间有限制 xff0c 在外面开板会花费好几天的制作和快递时间 xff0c 所以有时候就需要自己制版 xff0c 在这里我记录
Java打印九九乘法表

1 使用双重for循环打印九九乘法表 Java源代码如下 xff1a for int i 61 0 i lt 61 9 i 43 43 for int j 61 1 j lt 61 i j 43 43 System out print i
解决selenium打开Chrome浏览器自动退出的问题

好不容易安装好selenium和对应的浏览器驱动器后终于可以运行程序了 xff0c 结果发现一运行程序后浏览器打开就自动退出了 xff0c 但是我在Python代码中并没有写driver quit 方法 xff0c 上网查了查发现原来是我的
在Java应用中嵌入sshd服务

这个应用需要依赖apache mina的子项目sshd xff0c 项目主页http mina apache org sshd project index html xff0c 当前版本号为0 8 0 这里的sshd和Linux下的sshd
openssl开发库安装时的踩坑指南

序前几天用linux编译一个提权脚本的时候报错 openssl opensslv h 没有那个文件或目录的问题无论如何也解决不了 xff0c 这下我记录一个踩坑指南防止下一个人掉进坑里操作总体介绍首先介绍一下 xff0c 这个报
性能测试脚本用例【模板】

产品名称Product name 密级Confidentiality level 秘密产品版本Product version Total 12pages 共12页性能测试脚本用例仅供内部使用拟制日期 xff1a 审核日期 xff
Java常见的集合类

我们常见的Java集合类有List Set Map List 1 接口可以被继承 2 接口可以被多次实现 3 List和ArrayList package List import java util ArrayList import jav
WIN7我的电脑右键管理打不开

问题现象 xff1a 我的电脑右键点击管理无法正常打开 xff0c 会弹出下面的报错信息首先打开注册表 xff0c 打开运行 xff0c 输入regedit 选择路径 xff1a HKEY LOCAL MACHINE SOFTWARE C
LIKE的用法

我们来谈谈关于like运算符的理解 xff1a 下面是like的语法 xff0c 以后使用到like运算符的都必须根据这个语法使用 LIKE 运算符是用来匹配通配符指定模式的文本值如果搜索表达式与模式表达式匹配 xff0c LIKE 运算

随机推荐

从0开始详细安装archlinux（UEFI启动）

隔了一周没更新 xff0c 前阵子把电脑windows卸了装了个archlinux xff0c 不得不说arch是真的香 xff0c 但是坑也是真的多 xff0c 刚踩完所有的坑 xff0c 滚回来写blog了注 xff1a 本贴为UEF
archlinux开机无法联网问题，以及安装archlinuxcn和yay管理器

前一篇已经安装完了archlinux系统 xff0c 不过真正难的其实并不是安装 xff0c 你的路现在才开始 xff0c 哈哈添加用户建议 xff0c 不然使用登录管理器的时候不支持root用户 span class token fu
archlinux安装kde桌面和sddm登录管理器

前几篇已经配置好了archlinuxcn软件仓库 xff0c 网络和nvidia驱动 xff0c 现在来给你的archlinux安装一个kde桌面 xff08 kde玩法有很多 xff0c 可以自己去搜一搜美化教程 xff09 xff0c
Spring框架学习

目录目录学习内容 xff1a IoC java中创建对象有哪些方式 xff1a ioc的体现 xff1a DI 是ioc的技术实现 spring的第一个核心功能 ioc Spring 八大模块 Spring的特点 xff1a Sprin
CenOs6.7不能使用yum命令

因为官方不维护了所以先用更换源 wget O etc yum repos d CentOS Base repo https mirrors aliyun com repo Centos vault 6 10 repo https deve
PowerMock注解PowerMockIgnore的使用方法

故事要从一个异常开始 xff0c 某天我在开发一个加密解密特性 xff0c 算法使用的是3DES xff0c 样例代码如下 package org jackie study powermock import java io Unsuppo
如何写一棵AVL树

二叉查找树二叉查找树有一个缺陷就是查询效率跟树的高度有关在极端情况下 xff0c 查询效率为n 如何解决二叉查找树效率低问题 xff1f 要增加查询效率 xff0c 高效的方案是在插入的时候对树进行一下平衡操作 xff0c 降低树的高度
点击超链接下载.pdf文件

上代码 span class token keyword package span span class token class name Servlet span span class token punctuation span res
SpringTemplate增删改查及其事务控制基本使用

一 JdbcCRUD操作 1 导入坐标包 span class token generics span class token punctuation lt span dependency span class token punctuat
C语言从键盘读入一个正整数num，计算0 ~ num(包括num)范围内所有奇数之和

include span class token generics span class token punctuation lt span stdio span class token punctuation span h span cl
spring文件上传拦截器及异常处理

1文件上传当enctype 61 34 multipart form data 34 时request getParameter 方法失效单文件上传导入坐标 span class token generics span class t
jxl，java实现创建工作簿Excel并导入数据

导入jxl jar包只支持xls格式的excel 导入 span class token generics span class token punctuation lt span dependency span class token
springboot整合mybatis

先创建一个springbooot项目 file gt new project Spring Initalizr 勾选sql mysql Driver 这里和mybatis plus一起整合 span class token generics
Mybatis-Plus代码生成器

span class token operator lt span span class token operator span span class token operator span 代码生成器 span class token o
数据仓库Hive的分区表与分桶表

一分区表分区表实际上就是对应一个HDFS文件系统上的独立的文件夹 xff0c 该文件夹下是该分区所有的数据文件 Hive中的分区就是分目录 xff0c 把一个大的数据集根据业务需要分割成小的数据集在查询时通过WHERE子句中的表达式选
cloudflare动态域名解析-并实现自定义内外网混合动态域名DDNS脚本

1 获取CloudFlare的API令牌直接进入地址 xff1a API 令牌账户名 39 s Account Cloudflare 2 获取根域名ID curl X GET 34 https api cloudflare com cl
GitHub新手用法详解【适合新手入门-建议收藏！！！】

目录什么是Github xff0c 为什么使用它 xff1f 一 GitHub账号的注册与登录二 gitbash安装详解 1 git bash的下载与安装 2 git常用命令 3 Git 和 GitHub 的绑定 1 获取SSH key
小米mix2s刷win11和android双系统

在给电脑安装系统的过程中 xff0c 可能会因为各种原因出现windows无法安装的情况 xff0c 我在给小米mix2s安装win11时发现出现了计算机意外地重新启动或遇到错误 xff0c windows无法安装的情况 xff0c 下
Windows10开机引导文件被删除后系统修复

背景 xff1a 本人拯救者y9000p 自己装了个windows10的系统 xff0c 然后要用GAMMA xff0c 想装个Ubuntu的双系统来着 xff0c 但是装过gamma的朋友应该知道这玩意不好装 xff0c 装半天不成功不知
RabbitMQ快速入门及六大模式

目录核心组件运行原理实现步骤构建Gradle项目入门案例简单模式生产者 xff08 代码 xff09 消费者 xff08 代码 xff09 绑定交换机和队列发布订阅模式 xff08 Publish Subscribe xf

热门标签