RabbitMq之工作队列(轮询发送消息)

Posted 2023-01-29 微微笑再加油

工作队列(又称任务队列)的主要思想是避免立即执行资源密集型任务，而不得不等待它完成相反我们安排任务在之后执行。我们把任务封装为消息并将其发送到队列。在后台运行的工作进程将弹出任务并最终执行作业。当有多个工作线程时，这些工作线程将一起处理这些任务
1.创建一个连接mq工具类

package com.util;

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 Util 

    public static Channel getChannel() throws IOException, TimeoutException 
        ConnectionFactory factory = new ConnectionFactory();
        //连接ip 用户名 密码
        factory.setHost("192.168.126.130");
        factory.setUsername("root");
        factory.setPassword("123456");
        //创建连接
        Connection connection = factory.newConnection();
        //获取信道
        Channel channel = connection.createChannel();
        return channel;
    

2.创建多个工作线程（多个消费者）

package com.two;

import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.util.Util;

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

/**
 * 工作线程（相当于之前消费者）
 */
public class Work01 
    //队列名称
    public static final  String QUEUE_NAME ="hello";

    public static void main(String[] args) throws IOException, TimeoutException 
        Channel channel = Util.getChannel();

        //声明 接收消息
        DeliverCallback deliverCallback = (consumerTag, message) ->
            System.out.println("接收到的消息"+new String(message.getBody()));
        ;

        //取消消息时的回调
        CancelCallback cancelCallback = consumerTag ->
            System.out.println("消息消费被中断");
        ;
        System.out.println("第一个工作线程等待接收消息。。。。。。");
        //消息的接收
        channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);
    

注意：创建多个工作线程，只需要写如上一个，先运行

然后勾选如下选项

然后再改一下名字在运行

这样两个工作线程已经准备好了
3.准备生产者

package com.two;

import com.rabbitmq.client.Channel;
import com.util.Util;

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

/**
 * 生产者 发送大量的消息
 */
public class Task01 

    //队列名称
    public static final  String QUEUE_NAME ="hello";

    //发送大量消息
    public static void main(String[] args) throws IOException, TimeoutException 
        Channel channel = Util.getChannel();
        //生成一个队列
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //从控制台中接收消息
        Scanner scanner = new Scanner(System.in);
        while (scanner.hasNext())
            String message = scanner.next();
            channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
            System.out.println("发送消息完成"+message);
        
    

运行生产者，在控制台输入 aa bb cc dd

看两个工作线程接收到的消息如下


两个线程轮询分别获得对应四条数据