当前位置：首页 > news >正文

RabbitMQ工作模式

news 来源：原创 2024/11/17 2:04:03

🍁博客主页：👉不会压弯的小飞侠
✨欢迎关注：👉点赞👍收藏⭐留言✒
✨系列专栏：👉Linux专栏
🔥欢迎大佬指正，一起学习！一起加油！

在这里插入图片描述

🍁模式说明

工作队列（又名：任务队列）背后的主要思想是避免立即执行资源密集型任务，必须等待它要完成。相反，我们将任务安排在以后完成。我们将任务封装为消息并将其发送到队列。正在运行的工作进程在后台将弹出任务并最终执行工作。当您运行许多工作线程时，任务将在它们之间共享。
与入门程序的简单模式相比，多了一个或一些消费端，多个消费端共同消费同一个队列中的消息。
对于任务较多情况使用工作队列可以提高任务处理的速度。

🍁工作模式完成消息传递

在学习简单模式的时候工程已创建完成，只需要在不同包下测试即可。
- 需要创建一个工具类，将连接工厂，抽取出来。
- 在util包下创建ConnectionUtil类

public class ConnectionUtil {
    public static Connection getConnection() throws Exception {
        //定义连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //设置服务地址
        factory.setHost("**************");  //IP地址
        //端口
        factory.setPort(5672);
        //设置账号信息，用户名、密码、vhost
        factory.setVirtualHost("/");
        factory.setUsername("admin");
        factory.setPassword("root");
        // 通过工程获取连接
        Connection connection = factory.newConnection();
        return connection;
    }

    public static void main(String[] args) throws Exception {
        Connection con = ConnectionUtil.getConnection();
        System.out.println(con);
        con.close();
    }

}

编写生产者发送消息
- 编写消息生产者 Producter

public class Producer {
    static final String QUEUE_NAME = "work_queue";
    public static void main(String[] args) throws Exception {
        //获取连接
        Connection connection = ConnectionUtil.getConnection();
        //创建信道
        Channel channel = connection.createChannel();
        //声明队列
        channel.queueDeclare(QUEUE_NAME,true,false,false,null);
        for (int i = 1; i <= 10; i++) {
            String body = i+"Hello RabbitMQ!!!";
            channel.basicPublish("",QUEUE_NAME,null,body.getBytes());
        }
        channel.close();
        connection.close();
    }

}

编写消费者接收消息
- 编写消息消费者Consumer1

public class Consumer1 {
    static final String QUEUE_NAME = "work_queue";
    public static void main(String[] args) throws Exception {
        //获取连接
        Connection connection = ConnectionUtil.getConnection();
        //创建信道
        Channel channel = connection.createChannel();
        //创建Consumer对象
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        Consumer consumer = new DefaultConsumer(channel) {
            @Override   //唯一标识，交换机名称，
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("body：" + new String(body));
            }
        };
        //消费消息
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

编写消息消费者Consumer2

public class Consumer2 {
    static final String QUEUE_NAME = "work_queue";
    public static void main(String[] args) throws Exception {
        //获取连接
        Connection connection = ConnectionUtil.getConnection();
        //创建信道
        Channel channel = connection.createChannel();
        //创建Consumer对象
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        Consumer consumer = new DefaultConsumer(channel) {
            @Override   //唯一标识，交换机名称，
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("body：" + new String(body));
            }
        };
        //消费消息
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}