连接工具类：

public class rabbitMQUtils {
    //得到一个连接的 channel
    public static Channel getChannel() throws Exception{
        //1.创建一个连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.80.128");
        factory.setUsername("admin");
        factory.setPassword("123");
        //2.创建连接
        Connection connection = factory.newConnection();
        //3.创建频道
        Channel channel = connection.createChannel();
        return channel;
    }

一、简单模式

在下图中，“ P”是我们的生产者，“ C”是我们的消费者。中间的框是一个队列-RabbitMQ 代表使用者保留的消息缓冲区

消息产生者将消息放入队列
消息的消费者(consumer) 监听(while) 消息队列,如果队列中有消息,就消费掉,消息被拿走后,自动从队列中删除(隐患：消息可能没有被消费者正确处理,已经从队列中消失了,造成消息的丢失)

测试：

1.生产者

class producer implements Callable{
    private final static String QUEUE_NAME = "hello";

    @Override
    public Object call() throws Exception {
        //1.获取频道
        Channel channel = rabbitMQUtils.getChannel();
        //2.要发送的消息
        String msg="hello,RabbitMQ!";
        //3.生成一个队列
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //4.发送一个消息
        channel.basicPublish("",QUEUE_NAME,null,msg.getBytes(StandardCharsets.UTF_8));
        System.out.println("生产者发送了一条消息");
        return null;
    }
}

2.消费者

class consumer implements Callable {
    private final static String QUEUE_NAME = "hello";

    @Override
    public Object call() throws Exception {
        Channel channel = rabbitMQUtils.getChannel();
        System.out.println("消费者等待接收信息");
        //1.设置消费的回调消息
        DeliverCallback deliverCallback=(consumerTag, delivery)->{
            String message= new String(delivery.getBody());
            System.out.println("消费者收到了消息");
            System.out.println(message);
        };
        //2.取消消费的一个回调信息
        CancelCallback cancelCallback=(consumerTag)->{
            System.out.println("消息消费被中断");
        };
        //3.等待接收消息（参数2表示消费成功后自动应答 true-自动应答 false-手动应答）
        channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);
        return null;
    }
}

3.开启多线程

public static void main(String[] args) throws InterruptedException {
        ExecutorService service= Executors.newFixedThreadPool(10);
        service.submit(new consumer());
        Thread.sleep(3000);    //先让消费者做好接收信息的准备，然后生产者发送信息
        service.submit(new producer());
    }

//测试：
消费者等待接收信息
... ...
生产者发送了一条消息
消费者收到了消息
hello,RabbitMQ!

二、工作模式

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

2.1 轮询分发消息

任务会平均分配给各个线程。

测试：

1.消费者1

class consumer1 implements Callable {
    private final static String QUEUE_NAME = "hello";

    @Override
    public Object call() throws Exception {
        Channel channel = rabbitMQUtils.getChannel();
        System.out.println("消费者1等待接收信息");
        //1.设置消费的回调消息
        DeliverCallback deliverCallback=(consumerTag, delivery)->{
            String message= new String(delivery.getBody());
            System.out.println("消费者1收到了消息："+message);
        };
        //2.取消消费的一个回调信息
        CancelCallback cancelCallback=(consumerTag)->{
            System.out.println("消息消费被中断");
        };
        //3.等待接收消息（参数2表示消费成功后自动应答 true-自动应答 false-手动应答）
        channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);
        return null;
    }
}

2.消费者2

class consumer2 implements Callable {
    private final static String QUEUE_NAME = "hello";

    @Override
    public Object call() throws Exception {
        Channel channel = rabbitMQUtils.getChannel();
        System.out.println("消费者2等待接收信息");
        //1.设置消费的回调消息
        DeliverCallback deliverCallback=(consumerTag, delivery)->{
            String message= new String(delivery.getBody());
            System.out.println("消费者2收到了消息："+message);
        };
        //2.取消消费的一个回调信息
        CancelCallback cancelCallback=(consumerTag)->{
            System.out.println("消息消费被中断");
        };
        //3.等待接收消息（参数2表示消费成功后自动应答 true-自动应答 false-手动应答）
        channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);
        return null;
    }
}

3.生产者1

class producer1 implements Callable{
    private final static String QUEUE_NAME = "hello";

    @Override
    public Object call() throws Exception {
        //1.获取频道
        Channel channel = rabbitMQUtils.getChannel();
        //2.要发送的消息
        String msg1="1.吃饭";
        String msg2="2.睡觉";
        String msg3="3.打游戏";
        String msg4="4.谈恋爱";
        //3.生成一个队列
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //4.发送一个消息
        channel.basicPublish("",QUEUE_NAME,null,msg1.getBytes(StandardCharsets.UTF_8));
        System.out.println("生产者1发送了1条消息");
        channel.basicPublish("",QUEUE_NAME,null,msg2.getBytes(StandardCharsets.UTF_8));
        System.out.println("生产者1发送了1条消息");
        channel.basicPublish("",QUEUE_NAME,null,msg3.getBytes(StandardCharsets.UTF_8));
        System.out.println("生产者1发送了1条消息");
        channel.basicPublish("",QUEUE_NAME,null,msg4.getBytes(StandardCharsets.UTF_8));
        System.out.println("生产者1发送了1条消息");
        return null;
    }
}

4.开启多线程

 public static void main(String[] args) throws InterruptedException {
        ExecutorService service= Executors.newFixedThreadPool(10);
        service.submit(new consumer1());
        service.submit(new consumer2());

        Thread.sleep(2000);

        service.submit(new producer1());
    }

//测试：

消费者1等待接收信息
消费者2等待接收信息

生产者1发送了1条消息
生产者1发送了1条消息
生产者1发送了1条消息
生产者1发送了1条消息

消费者2收到了消息：1.吃饭
消费者2收到了消息：3.打游戏
消费者1收到了消息：2.睡觉
消费者1收到了消息：4.谈恋爱

通过程序执行发现生产者总共发送 4 个消息，消费者 1 和消费者 2 分别分得两个消息，并且是按照有序的一个接收一次消息。

2.2 消息应答

RabbitMQ 一旦向消费者传递了一条消息，便立即将该消息标记为删除。在这种情况下，突然有个消费者挂掉了，我们将丢失正在处理的消息以及后续发送给该消费者的消息，因为它无法接收到。

为了保证消息在发送过程中不丢失，rabbitmq 引入消息应答机制，消息应答就是消费者在接收到消息并且处理该消息之后，告诉 rabbitmq 它已经处理了，rabbitmq 可以把该消息删除了。

1.自动应答，是消息发送后立即被认为传送成功，这种模式需要在高吞吐量和数据传输安全性方面做权衡，会有消息丢失的风险。

2.手动应答，手动应答的好处是可以批量应答并且减少网络拥堵。常用方法：

Channel.basicAck(DeliveryTag,false)
//(用于肯定确认)已知道并且成功处理了消息，可以丢弃了
//第二个参数multiple：为false表示通知 RabbitMQ 当前消息被确认
//为true则额外将比第一个参数指定的 delivery tag 小的消息一并确认（批量）

channel.basicReject(deliveryTag, true);
//basic.reject方法拒绝deliveryTag对应的消息，
//第二个参数是否重新入队，true重新入队，false丢弃或者进入死信队列。
//该方法reject后，该消费者还是会消费到该条被reject的消息。

channel.basicNack(deliveryTag, false, true);
//不确认deliveryTag对应的消息，第二个参数multiple：是否应用于多消息，
//第三个参数是否重新入队，与basic.reject区别就是同时支持多个消息，
//可以nack该消费者先前接收未ack的所有消息。nack后的消息也会被自己消费到

关于multiple：

true 代表批量应答 channel 上未应答的消息。比如说 channel 上有传送 tag 的消息 5,6,7,8 当前 tag 是 8 那么此时 5-8 的这些还未应答的消息都会被确认收到消息应答

false 则只会应答 tag=8 的消息，5,6,7 这三个消息依然不会被确认收到消息应答

如果消费者由于某些原因失去连接(其通道已关闭，连接已关闭或 TCP 连接丢失)，导致消息 未发送 ACK 确认，RabbitMQ 将了解到消息未完全处理，并将对其重新排队。如果此时其他消费者可以处理，它将很快将其重新分发给另一个消费者。这样，即使某个消费者偶尔死亡，也可以确保不会丢失任何消息。

默认消息采用的是自动应答，所以我们要想实现消息消费过程中不丢失，需要把自动应答改为手动应答。

修改应答方式要修改消费者的代码，如下：

class consumer1 implements Callable {
    private final static String QUEUE_NAME = "hello";

    @Override
    public Object call() throws Exception {
        Channel channel = rabbitMQUtils.getChannel();
        System.out.println("消费者1等待接收信息");
        //1.设置消费的回调消息
        DeliverCallback deliverCallback=(consumerTag, delivery)->{
            String message= new String(delivery.getBody());
            System.out.println("消费者1收到了消息："+message);
        //***手动应答确认收到***
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
        };
        //2.取消消费的一个回调信息
        CancelCallback cancelCallback=(consumerTag)->{
            System.out.println("消息消费被中断");
        };
        //3.等待接收消息（参数2表示消费成功后自动应答 true-自动应答 false-手动应答）
        channel.basicConsume(QUEUE_NAME,false,deliverCallback,cancelCallback);
        return null;
    }
}

2.3 RabbitMQ 持久化

刚刚我们已经看到了如何处理任务不丢失的情况，但是如何保障当 RabbitMQ 服务停掉以后消息生产者发送过来的消息不丢失。默认情况下 RabbitMQ 退出或由于某种原因崩溃时，会忽视队列和消息。确保消息不会丢失需要做两件事：我们需要将队列和消息都标记为持久化。

1.如何实现队列持久化？

在声明队列时把durable参数设置为持久化：

boolean durable=true；
//第二个参数是设置持久化的
channel.queueDeclare(QUEUE_NAME,durable,false,false,null);

注：如果之前声明的队列不是持久化的，需要把原先队列先删除，或者重新创建一个持久化的队列，不然就会出现错误

2.如何实现消息持久化？

在消息生产者修改准备发布代码，添加MessageProperties.PERSISTENT_TEXT_PLAIN 属性。

//第三个参数是将消息持久化
channel.basicPublish("",QUEUE_NAME,MessageProperties.PERSISTENT_TEXT_PLAIN,
msg.getBytes(StandardCharsets.UTF_8));

注：将消息标记为持久化并不能完全保证不会丢失消息。尽管它告诉 RabbitMQ 将消息保存到磁盘，但是这里依然存在当消息刚准备存储在磁盘的时候但是还没有存储完，消息还在缓存的一个间隔点。此时并没有真正写入磁盘。持久性保证并不强，但是对于我们的简单任务队列而言，这已经绰绰有余了。

2.4 不公平分发

在最开始学习到 RabbitMQ分发消息采用轮询分发，但是在某种场景下这种策略并不是很好，比方说有两个消费者在处理任务，其中有个消费者 1 处理任务的速度非常快，而另外一个消费者 2 处理速度却很慢，这个时候我们还是采用轮询分发的话就会使得处理速度快的消费者大部分时间处于空闲状态，而处理慢的那个消费者一直在干活。（作为资本家，我想榨干员工的所有价值）

为了避免这种情况，可以在消费者中设置参数：

//参数是prefetchCount
channel.basicQos(1);

如果这个任务我还没有处理完或者我还没有应答你，你先别分配给我，我目前只能处理一个任务，然后 rabbitmq 就会把该任务分配给没有那么忙的那个空闲消费者，当然如果所有的消费者都没有完成手上任务，队列还在不停的添加新任务，队列有可能就会遇到队列被撑满的情况，这个时候就只能添加新的 worker 或者改变其他存储任务的策略。

prefetchCount定义了该通道上允许的未确认消息的最大数量，一旦数量达到配置的数量， RabbitMQ 将停止在通道上传递更多消息，除非至少有一个未处理的消息被确认。

例如，假设在通道上有未确认的消息 5、6、7，8，并且通道的预取计数设置为 4，此时 RabbitMQ 将不会在该通道上再传递任何消息，除非至少有一个未应答的消息被 ack。比方说 tag=6 这个消息刚刚被确认 ACK，RabbitMQ 将会感知这个情况到并再发送一条消息。

消息应答和 QoS 预取值对用户吞吐量有重大影响。通常，增加预取将提高向消费者传递消息的速度。虽然自动应答传输消息速率是最佳的，但是，在这种情况下已传递但尚未处理的消息的数量也会增加，从而增加了消费者的 RAM 消耗。应该小心使用具有无限预处理的自动确认模式或手动确认模式。

2.5 发布确认

生产者将信道设置成 confirm 模式，一旦信道进入 confirm 模式，所有在该信道上面发布的消息都将会被指派一个唯一的 ID(从 1 开始)，一旦消息被投递到所有匹配的队列之后，broker 就会发送一个确认给生产者(包含消息的唯一 ID)，这就使得生产者知道消息已经正确到达了目的队列，如果消息和队列是可持久化的，那么确认消息会在将消息写入磁盘之后发出，broker 回传给生产者的确认消息中 delivery-tag 域包含了确认消息的序列号，此外 broker 也可以设置 basic.ack 的 multiple 域，表示到这个序列号之前的所有消息都已经得到了处理。（这是broker操作的，跟应答有区别）

confirm 模式最大的好处在于他是异步的，一旦发布一条消息，生产者应用程序就可以在等信道返回确认的同时继续发送下一条消息，当消息最终得到确认之后，生产者应用便可以通过回调方法来处理该确认消息，如果 RabbitMQ 因为自身内部错误导致消息丢失，就会发送一条 nack 消息，生产者应用程序同样可以在回调方法中处理该 nack 消息。

如何开启发布确认？

调用生产者的 channel.confirmSelect() 方法。

2.5.1 同步确认发布

这是一种简单的确认方式，就是发布一个消息之后只有它被确认发布，后续的消息才能继续发布。这种确认方式有一个最大的缺点就是:发布速度特别的慢，因为如果没有确认发布的消息就会阻塞所有后续消息的发布。

测试：

@Test
public void testSyncPublishConfirm() throws Exception {
        Channel channel = rabbitMQUtils.getChannel();
        //开启发布确认
        channel.confirmSelect();
        //发送1000条消息，内容为0-999
        for(int i=0;i<1000;i++){
            channel.basicPublish("","test",null,String.valueOf(i).getBytes());
            //发送每条消息都要等待它的确认信息
            boolean flag = channel.waitForConfirms();
            if(!flag) System.out.println("消息发送失败！");
            else System.out.println("消息发送成功！");
    }
}

这种方式很耗时，我们也可以先发布一批消息然后一起确认以提高吞吐量，也就是批量确认发布，但当发生故障导致发布出现问题时，不知道是哪个消息出现问题了，我们就必须将整个批处理保存在内存中，以记录重要的信息而后重新发布消息。

2.5.2 异步确认发布（推荐）

异步确认虽然编程逻辑比上两个要复杂，但是性价比最高，无论是可靠性还是效率都没得说，它是利用回调函数来达到消息可靠性传递的，这个中间件也是通过函数回调来保证是否投递成功。

生产者发送的每一条消息都有一个编号，在发送过程中会有一个异步的线程来监听消息是否发送成功，并且可以记录发送成功、与失败的信息，回传给生产者。

如何实现？

使用确认监听器-channel.addConfirmListener（）

它有两种参数：

我用第一种参数（匿名内部类）来实现：

//还是生产者
class ASyncPublishConfirm implements Callable{
    private final static String QUEUE_NAME = "hello";
    @Override
    public Object call() throws Exception {
        Channel channel = rabbitMQUtils.getChannel();
        //声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        //开启发布确认
        channel.confirmSelect();
        //开启确认监听器
        channel.addConfirmListener(new ConfirmListener() {
            @Override
            public void handleAck ( long deliveryTag, boolean multiple) throws IOException {
                System.out.println("ack:"+deliveryTag);
            }
            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                System.out.println("default:"+deliveryTag);
            }
        });
        //发送100条消息，内容为0-999
        for(int i = 0;i<1000;i++){
            channel.basicPublish("", QUEUE_NAME, null, String.valueOf(i).getBytes(StandardCharsets.UTF_8));
        }
        return null;
    }
}

消费者：

class consumer3 implements Callable {
    private final static String QUEUE_NAME = "hello";

    @Override
    public Object call() throws Exception {
        Channel channel = rabbitMQUtils.getChannel();
        System.out.println("消费者1等待接收信息");
        //1.设置消费的回调消息
        DeliverCallback deliverCallback=(consumerTag, delivery)->{
            String message= new String(delivery.getBody());
            System.out.println("消费者1收到了消息："+message);
            //手动应答确认收到
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
        };
        //2.取消消费的一个回调信息
        CancelCallback cancelCallback=(consumerTag)->{
            System.out.println("消息消费被中断");
        };
        //3.等待接收消息（参数2表示消费成功后自动应答 true-自动应答 false-手动应答）
        channel.basicConsume(QUEUE_NAME,false,deliverCallback,cancelCallback);
        return null;
    }
}

多线程测试：

 public static void main(String[] args) throws InterruptedException{
        ExecutorService service= Executors.newFixedThreadPool(10);
        service.submit(new consumer3());
        Thread.sleep(2000);
        service.submit(new ASyncPublishConfirm());
}

//部分结果：
消费者1收到了消息：912
ack:996
消费者1收到了消息：913
ack:997
ack:999
消费者1收到了消息：914
ack:1000
消费者1收到了消息：915
消费者1收到了消息：916
消费者1收到了消息：917
... ...
消费者1收到了消息：999

本文含有隐藏内容，请开通VIP 后查看

RabbitMQ-简单模式/工作模式（分发、应答、持久化、不公平分发、发布确认）

一、简单模式

二、工作模式

2.1 轮询分发消息

2.2 消息应答

2.3 RabbitMQ 持久化

2.4 不公平分发

2.5 发布确认

2.5.1 同步确认发布

2.5.2 异步确认发布（推荐）

网站公告

今日签到

热门文章

最新发布