Java真的不难（五十四）RabbitMQ的入门及使用

阿里云国内75折 回扣 微信号：monov8
阿里云国际，腾讯云国际，低至75折。AWS 93折 免费开户实名账号 代冲值 优惠多多 微信号：monov8 飞机：@monov6

RabbitMQ的入门及使用

一、什么是RabbitMQ

MQ全称为Message Queue即消息队列。消息队列是在消息的传输过程中保存消息的容器。它是典型的生产者、消费者模型。生产者不断向消息队列中生产消息消费者不断的从队列中获取消息。因为消息的生产和消费都是异步的而且只关心消息的发送和接收没有业务逻辑的侵入这样就实现了生产者和消费者的解耦

二、RabbitMQ与Kafka的全面对比

对比项KafkaRabbitMQ
开发语言scala、Javaerlang
是否支持多租户2.x.x支持支持
是否支持topic优先级不支持支持
是否支持消息全局有序不支持支持
是否支持消息分区有序支持支持
是否有内置监控无有
是否支持多个生产者支持支持
是否支持多个消费者支持支持
是否支持一个分区多个消费者不支持不支持
是否支持JMX支持不支持
是否支持加密支持支持
消息队列协议支持仅支持自定义协议支持AMQP、MQTT、STOMP协议
客户端语言支持支持支持
是否支持消息追踪不支持支持
是否支持消费者推模式不支持支持
否支持消费者拉模式支持支持
是否支持广播消息支持支持
是否支持消息回溯支持消息回溯因为消息持久化消息被消费后会记录offset和timstamp不支持消息确认被消费后会被删除
是否支持消息数据持久化支持支持
是否支持流量控制支持支持
是否支持事务性消息支持不支持
元数据管理通过zookeeper进行管理支持
默认服务端口90925672
默认监控端口kafka web console 9000;kafka manager 9000;15672
相对网络开销较小较大
相对内存消耗较小较大
相对cpu消耗较大较小

实际场景选择

Kafka
常作为消息传输的数据管道 优势主要体现在吞吐量上虽然可以通过策略实现数据不丢失严谨性上不如 RabbitMQ但 kafka保证每条消息最少送达一次有较小的概率会出现数据重复发送的情况 若消息吞吐量极大则Kafka

RabbitMQ
RabbitMQ金融场景中经常使用 常作为交易数据作为数据传输管道 具有较高的严谨性数据丢失的可能性更小具备更高的实时性和Spring是统一厂商开发后期支持比较好目前最流行的对容错性的处理比较完善

RabbitMQ 支持发布订阅、轮询分发、公平分发、重发、消息拉取
Kafka 不支持重发、事务

三、Linux上安装RabbitMQ

这次安装将RabbitMQ部署在Linux上可以在电脑本地安装一台Linux也可以购买云服务器若购买云服务器则需要在安全组内把后面要用到的端口给打开

1、使用Docker安装最简单的方法

拉取RabbitMQ

以下均为Linux指令

#拉取RabbitMQ
docker pull RabbitMQ

#启动rabbitmq
docker run -d --hostname my-rabbit --name rabbit -p 15672:15672 -p 5672:5672 rabbitmq

#查看docker目前在运行的容器是否有rabbitmq
docker ps

2、图形化安装插件

#进入运行中的容器
docker exec -it 镜像ID /bin/bash

#rabbitmq图形化安装插件
rabbitmq-plugins enable rabbitmq_management

3、WEB页面开启资源监控

#进入容器
docker exec -it rabbitmq /bin/bash

#切到对应目录
cd /etc/rabbitmq/conf.d/

#修改 management_agent.disable_metrics_collector = false
echo management_agent.disable_metrics_collector = false > management_agent.disable_metrics_collector.conf

#退出容器
exit

#重启容器
docker restart rabbitmq

然后在浏览器内输入服务器地址+端口15672即可进入WEB管理页面默认账号密码均为guest

在这里插入图片描述

四、Docker配置RabbitMQ集群

先停止在运行的MQ

docker stop 运行容器ID

#启动三个容器
docker run -d --hostname rabbitmq01 --name rabbitmqCluster01 -p 15672:15672 -p 5672:5672 -p 1883:1883 -e RABBITMQ_ERLANG_COOKIE='rabbitmqCookie' rabbitmq

docker run -d --hostname rabbitmq02 --name rabbitmqCluster02 -p 15673:15672 -p 5673:5672 -p 1884:1883 -e RABBITMQ_ERLANG_COOKIE='rabbitmqCookie' --link rabbitmqCluster01:rabbitmq01 rabbitmq

docker run -d --hostname rabbitmq03 --name rabbitmqCluster03 -p 15674:15672 -p 5674:5672 -p 1885:1883 -e RABBITMQ_ERLANG_COOKIE='rabbitmqCookie' --link rabbitmqCluster01:rabbitmq01 --link rabbitmqCluster02:rabbitmq02 rabbitmq

#Erlang Cookie 值必须相同也就是一个集群内 RABBITMQ_ERLANG_COOKIE 参数的值必须相同。因为 RabbitMQ 是用Erlang实现的Erlang Cookie 相当于不同节点之间通讯的密钥Erlang节点通过交换 Erlang Cookie 获得认证


#进入第二个容器
docker exec -it rabbitmqCluster02 bash
rabbitmqctl stop_app
rabbitmqctl reset

#加入集群
rabbitmqctl join_cluster --ram rabbit@rabbitmq01
rabbitmqctl start_app
exit


#进入第三个容器
docker exec -it rabbitmqCluster03 bash
rabbitmqctl stop_app
rabbitmqctl reset

#加入集群
rabbitmqctl join_cluster --ram rabbit@rabbitmq01
rabbitmqctl start_app
exit

#主要参数
-p 15672:15672 management 界面管理访问端口
-p 5672:5672 amqp 访问端口
-p 1883:1883 mqtt  访问端口

然后依次运行2、3的步骤安装插件和开启资源监控
上诉15672、15673、15674端口均需要在服务器安全组内开启

完成后即可的WEB页面看到MQ的集群

五、SpringBoot整合使用RabbitMQ

依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

yml文件的配置

spring:
  rabbitmq:
    username: guest
    password: guest
    virtual-host: /
    host: 通道运行的地址
    port: 5672

生产者配置类Config

模拟美团外卖下单正常下单后若马上被接单则被这条信息直接消费若5秒内没人接单该订单消息将进入加急派单队列死信队列

所以我们需要两台交换机和两个通道

@Configuration
public class mtRabbitConfig {

	//正常通道的交换机
    @Bean
    public FanoutExchange mtExchange(){
        return new FanoutExchange("mt_fanout_exchange",true,false);
    }

	//死信通道的交换机
    @Bean
    public FanoutExchange mtDeadExchange(){
        return new FanoutExchange("mt_fanout_dead_exchange",true,false);
    }

	//正常通道给消息设计过期时间超过该时间未被消费则进入指定的mt_fanout_dead_exchange
    @Bean
    public Queue mtQueue(){
        Map<String,Object> args = new HashMap<>();
        args.put("x-message-ttl",5000);
        args.put("x-dead-letter-exchange","mt_fanout_dead_exchange");
        return new Queue("mt_queue",true,false,false,args);
    }

    @Bean
    public Queue mtDaedQueue(){
        return new Queue("mt_dead_queue",true,false,false);
    }

	//交换机与通道绑定
    @Bean
    public Binding mtBinding(){
        return BindingBuilder.bind(mtQueue()).to(mtExchange());
    }

    @Bean
    public Binding mtDeadBinding(){
        return BindingBuilder.bind(mtDaedQueue()).to(mtDeadExchange());
    }
}

生产者模拟用户下单

    @Autowired
    private RabbitTemplate rabbitTemplate;

    //模拟美团订单下单若5秒不接单消费,则进入死信队列加急派单
    public void mtTakeOutOrder(String name, String food, String number) {
        UUID takeOutId = UUID.randomUUID();
        String orderTime = DateFormat.getDateTimeInstance().format(new Date());
        String exchangeName = "mt_fanout_exchange";
        String takeOutMes = "美团订单编号" + takeOutId + " " + orderTime + " " + name + " " + food + " " + number;
        String routingKey = "";
        
		//将消息放入通道内
        Object result = rabbitTemplate.convertSendAndReceive(exchangeName, routingKey, takeOutMes);
        System.out.println("配送中心响应"+result);
    }

测试类

    @Test
    void mtTakeOutOrder() throws InterruptedException {
         takeOutOrder.mtTakeOutOrder("小张"+i, "麻辣烫", "10086");
 }

消费者外卖接单中心

正常接单消费者

@RabbitListener(bindings = @QueueBinding(value = @Queue(value = "mt_queue", autoDelete = "false"),
        exchange = @Exchange(value = "mt_fanout_exchange", type = ExchangeTypes.FANOUT)))
@Component
public class mtTakeOutDelivery {

    @RabbitHandler
    public String buyTrainTickets(String message) {
        System.out.println("正常美团外卖订单已接单" + message);
        return "配送中心已接单";
    }
}

加急接单消费者死信队列内的消息

@RabbitListener(bindings = @QueueBinding(value = @Queue(value = "mt_dead_queue", autoDelete = "false"),
        exchange = @Exchange(value = "mt_fanout_dead_exchange", type = ExchangeTypes.FANOUT)))
@Component
public class mtDeadTakeOutDelivery {

    @RabbitHandler
    public String buyTrainTickets(String message) {
        System.out.println("加急饿了么外卖订单已接单" + message);
        return "配送中心已接单";
    }
}

消费者运行后只要监听的两个通道内有消息就会被消费

六、RabbitMQ的手动ACK

为了确保消息不会丢失RabbitMQ支持消息应答。消费者发送一个消息应答告诉RabbitMQ这个消息已经接收并且处理完毕了。RabbitMQ就可以删除它了。

自动应答就是上面的案例只要被消费者取出通道内就会删除这个消息万一这个消息在消费者那边处理异常因为通道里已经没用这条消息了就会出现消息丢失。所以在有些场景需要改为手动应答ACK就是消费者把这条消息确认处理完毕后再告诉通道删除消息若异常这条消息将返回通道内可以重新处理这就是手动应答。

还是一样来配置两个通道

生产者配置

@Configuration
public class TestQueueConfig {


    @Bean
    public FanoutExchange TestExchange() {
        return new FanoutExchange("test_exchange", true, false);
    }

    @Bean
    public FanoutExchange TestDeadExchange() {
        return new FanoutExchange("test_dead_exchange", true, false);
    }

    @Bean
    public Queue TestDeadQueue() {
        return new Queue("test_dead_queue", true, false, false);
    }

    @Bean
    public Queue TestQueue() {
        Map<String, Object> args = new HashMap<>();
        //20秒钟未消费转到死信队列
        args.put("x-message-ttl", 5000);
        args.put("x-dead-letter-exchange", "test_dead_exchange");
        return new Queue("test_queue", true, false, false, args);
    }

    @Bean
    public Binding TestBinding() {
        return BindingBuilder.bind(TestQueue()).to(TestExchange());
    }

    @Bean
    public Binding TestDeadBinding() {
        return BindingBuilder.bind(TestDeadQueue()).to(TestDeadExchange());
    }
}

生产者业务代码不变与上个案例一致也行

消费者配置类

@Configuration
public class MyselfReceiverConfig {

    @Autowired
    private CachingConnectionFactory cachingConnectionFactory;

    @Autowired
    private MyselfReceiver myselfReceiver;

    @Autowired
    private MyselfDeadReceiver myselfDeadReceiver;

    @Bean
    public SimpleMessageListenerContainer simpleMessageListenerContainer(){
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(cachingConnectionFactory);
        container.setConcurrentConsumers(1);
        container.setMaxConcurrentConsumers(10);
        //手动确认
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        container.setQueueNames("test_queue");
        container.setMessageListener(myselfReceiver);
        return container;
    }

    @Bean
    public SimpleMessageListenerContainer simpleMessageListenerContainer_Dead(){
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(cachingConnectionFactory);
        container.setConcurrentConsumers(1);
        container.setMaxConcurrentConsumers(10);
        //手动确认
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        container.setQueueNames("test_dead_queue");
        container.setMessageListener(myselfDeadReceiver);
        return container;
    }
}

正常消费者

@Component
public class MyselfReceiver implements ChannelAwareMessageListener {
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            byte[] bytes = message.getBody();
            String mes = new String(bytes);
            String substring = mes.replace("\\\"","'");
            System.out.println("正常通道内消息"+substring);
            //业务主体

            //若业务处理无异常则回复通道删除消息
            channel.basicAck(deliveryTag,true);
        }catch (Exception e){
            channel.basicReject(deliveryTag,true);
        }
    }
}

死信通道消费者

@Component
public class MyselfDeadReceiver implements ChannelAwareMessageListener {
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            byte[] bytes = message.getBody();
            String mes = new String(bytes);
            String result = mes.replace("\\\"", "'");
            System.out.println("死信通道内消息" + result);
            //业务主体

            //若业务处理无异常则回复通道删除消息
            channel.basicAck(deliveryTag, true);
        } catch (Exception e) {
            //有异常把消息返回通道
            channel.basicReject(deliveryTag, true);
        }
    }
}

这样就完成了手动ACK若消费者处理没有异常将使用channel.basicAck(deliveryTag, true);
若出现了异常将使用channel.basicReject(deliveryTag, true);此消息将重新进入通道内这也确保了未消费成功的消息不会出现丢失的情况。

在这里插入图片描述

阿里云国内75折 回扣 微信号：monov8
阿里云国际，腾讯云国际，低至75折。AWS 93折 免费开户实名账号 代冲值 优惠多多 微信号：monov8 飞机：@monov6
标签: JavaRabbitMQ
返回列表

上一篇：网络基础-基础网络命令

下一篇：SpringBoot自动装配原理

“Java真的不难（五十四）RabbitMQ的入门及使用” 的相关文章

【数据库课设】机票预订系统 java+mysql实现 附源码_机票预订系统1年前 (2023-02-02)
[Java进阶] Swing两万字大总结一（超详细教程，这不得收藏一波）1年前 (2023-02-02)
Java中的多线程安全问题1年前 (2023-02-02)
Java学习（86）Java集合——案例：宠物猫信息管理（HashSet增删改查）1年前 (2023-02-02)
【JavaSE】异常的初步认识1年前 (2023-02-02)
Java学习（84）Java集合——案例：公告管理（ArrayList增删改查）1年前 (2023-02-02)
《SpringBoot篇》23.SpringBoot整合JavaMail实现发送邮件详解_springboot整合javamail1年前 (2023-02-02)
Java之集合概述、ArrayList的使用、常见API和集合案例1年前 (2023-02-02)
Java基础之 Integer 类源码分析1年前 (2023-02-02)
面试题-java高级（答案超详细）1年前 (2023-02-02)

阿里云国际版