服务器的异步通信——RabbitMQ2

简介: 服务器的异步通信——RabbitMQ

服务器的异步通信——RabbitMQ1:https://developer.aliyun.com/article/1521829

工作消息队列(WorkQueue)

下面场景中如果queue中有50条请求消息,但是consumer1只能处理40条,剩余的10条就可以由consumer进行处理,所以说工作消息队列可以提高消息的处理速度,避免队列消息堆积


模拟Workqueue,实现一个队列绑定多个消费者,基本实现思路如下:

  1. 在publisher服务中定义测试方法,每秒产生50条消息,发送到simple.queue中
  2. 在consumer服务中定义两个消息监听者,都监听simple.queue队列
  1. 消费者1每秒处理50条消息,消费者2每秒处理10条消息

代码实现:

在publisher服务中定义测试方法,每秒产生50条消息,发送到simple.queue中

    public void testSendMessage2WorkQueue() throws InterruptedException {
        String queueName = "simple.queue";
        String message = "hello, message__";
        for (int i = 1; i <= 50; i++) {
            rabbitTemplate.convertAndSend(queueName, message + i);
            Thread.sleep(20);
        }
    }

在consumer服务中定义两个消息监听者,都监听simple.queue队列,设置消费者1每秒处理50条消息,消费者2每秒处理10条消息

    @RabbitListener(queues = "simple.queue")
    public void listenWorkQueue1(String msg) throws InterruptedException {
        System.out.println("消费者1接收到消息:【" + msg + "】" + LocalTime.now());
        Thread.sleep(20);
    }
 
    @RabbitListener(queues = "simple.queue")
    public void listenWorkQueue2(String msg) throws InterruptedException {
        System.err.println("消费者2........接收到消息:【" + msg + "】" + LocalTime.now());
        Thread.sleep(200);
    }

修改application.yml文件,设置preFetch这个值,可以控制预取消息的上限,确保消费者2取消息时只能取一条,提高效率(“能者多劳”):

spring:
  rabbitmq:
    listener:
      simple:
        prefetch: 1

运行结果:

发布、订阅(Publish、Subscribe)

发布订阅模式与之前案例的区别就是允许将同一消息发送给多个消费者。实现方式是加入了exchange(交换机)。

常见exchange类型包括:

  • Fanout:广播
  • Direct:路由
  • Topic:话题

exchange负责消息路由,而不是存储,路由失败则消息丢失

Fanout Exchange

Fanout Exchange会将接收到的消息路由到每一个跟其绑定的queue中,如下:

基本实现思路如下:

  1. 在consumer中,利用代码声明队列、交换机,将二者进行绑定
  2. 在consumer中,编写两个消费方法,分别监听fanout.queue1和fanout.queue2
  1. 在publisher中编写测试方法,向fanout发送消息

代码实现:

在consumer中,利用代码声明队列、交换机,将二者进行绑定

@Configuration
public class FanoutConfig {
    // itcast.fanout
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("itcast.fanout");
    }
 
    // fanout.queue1
    @Bean
    public Queue fanoutQueue1(){
        return new Queue("fanout.queue1");
    }
 
    // 绑定队列1到交换机
    @Bean
    public Binding fanoutBinding1(Queue fanoutQueue1, FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
    }
 
    // fanout.queue2
    @Bean
    public Queue fanoutQueue2(){
        return new Queue("fanout.queue2");
    }
 
    // 绑定队列2到交换机
    @Bean
    public Binding fanoutBinding2(Queue fanoutQueue2, FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
    }
}

在consumer中,编写两个消费方法,分别监听fanout.queue1和fanout.queue2

    @RabbitListener(queues = "fanout.queue1")
    public void listenFanoutQueue1(String msg) {
        System.out.println("消费者接收到fanout.queue1的消息:【" + msg + "】");
    }
    @RabbitListener(queues = "fanout.queue2")
    public void listenFanoutQueue2(String msg) {
        System.out.println("消费者接收到fanout.queue2的消息:【" + msg + "】");
    }

在publisher中编写测试方法,向fanout发送消息

    @Test
    public void testSendFanoutExchange() {
        // 交换机名称
        String exchangeName = "itcast.fanout";
        // 消息
        String message = "hello, every one!";
        // 发送消息
        rabbitTemplate.convertAndSend(exchangeName, "", message);
    }

运行结果:

Direct Exchange

Direct Exchange会将接收到的消息根据规则路由到指定的Queue,因此被称为路由模式

  • l每一个Queue都与Exchange设置一个BindingKey
  • l发布者发送消息时,指定消息的RoutingKey
  • lExchange将消息路由到BindingKey与消息RoutingKey一致的队列

基本实现思路如下:

  1. 利用@RabbitListener声明ExchangeQueueRoutingKey
  2. consumer服务中,编写两个消费者方法,分别监听direct.queue1direct.queue2
  3. publisher中编写测试方法,向itcast. direct发送消息


代码实现:

在consumer服务中,编写两个消费者方法,分别监听direct.queue1direct.queue2,并利用@RabbitListener声明ExchangeQueueRoutingKey

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue1"),
            exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),
            key = {"red", "blue"}
    ))
    public void listenDirectQueue1(String msg){
        System.out.println("消费者接收到direct.queue1的消息:【" + msg + "】");
    }
 
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue2"),
            exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),
            key = {"red", "yellow"}
    ))
    public void listenDirectQueue2(String msg){
        System.out.println("消费者接收到direct.queue2的消息:【" + msg + "】");
    }

在publisher服务发送消息到DirectExchange

    @Test
    public void testSendDirectExchange() {
        // 交换机名称
        String exchangeName = "itcast.direct";
        // 消息
        String message = "hello, red!";
        // 发送消息
        rabbitTemplate.convertAndSend(exchangeName, "red", message);
    }

运行结果:

Topic Exchange

Topic Exchange与Direct Exchange类似,区别在于Topic Exchange的routingKey必须是多个单词的列表,并且以.分割

QueueExchange指定BindingKey时可以使用通配符:

#:代指0个或多个单词

*:代指一个单词

基本实现思路如下:

  1. 利用@RabbitListener声明ExchangeQueueRoutingKey
  2. 在consumer服务中,编写两个消费者方法,分别监听topic.queue1topic.queue2
  1. 在publisher中编写测试方法,向itcast. topic发送消息

代码实现:

利用@RabbitListener声明ExchangeQueueRoutingKey,在consumer服务中,编写两个消费者方法,分别监听topic.queue1topic.queue2

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "topic.queue1"),
            exchange = @Exchange(name = "itcast.topic", type = ExchangeTypes.TOPIC),
            key = "china.#"
    ))
    public void listenTopicQueue1(String msg){
        System.out.println("消费者接收到topic.queue1的消息:【" + msg + "】");
    }
 
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "topic.queue2"),
            exchange = @Exchange(name = "itcast.topic", type = ExchangeTypes.TOPIC),
            key = "#.news"
    ))
    public void listenTopicQueue2(String msg){
        System.out.println("消费者接收到topic.queue2的消息:【" + msg + "】");
    }

在publisher中编写测试方法,向itcast. topic发送消息

    @Test
    public void testSendTopicExchange() {
        // 交换机名称
        String exchangeName = "itcast.topic";
        // 消息
        String message = "今天天气不错,我的心情好极了!";
        // 发送消息
        rabbitTemplate.convertAndSend(exchangeName, "china.weather", message);
    }

运行结果:

SpringAMQP-消息转换器

SpringAMQP的发送方法中,接收消息的类型是Object,也就是说我们可以发送任意对象类型的消息,SpringAMQP会帮我们序列化为字节后发送。

Spring的对消息对象的处理是由org.springframework.amqp.support.converter.MessageConverter来处理的。而默认实现是SimpleMessageConverter,基于JDK的ObjectOutputStream完成序列化。


如果要修改只需要定义一个MessageConverter 类型的Bean即可。


推荐用JSON方式序列化,实现步骤如下:

在父工程中引入依赖

        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
        </dependency>

在publisher和consumer服务中声明MessageConverter:

    @Bean
    public MessageConverter messageConverter(){
        return new Jackson2JsonMessageConverter();
    }

相关实践学习
消息队列RocketMQ版:基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
目录
相关文章
|
13天前
|
消息中间件 测试技术 RocketMQ
消息队列 MQ产品使用合集之在异步发送消息函数sendMessage()中出现了错误,错误代码为-3,该如何解决
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
|
22天前
|
Java
Java Socket编程与多线程:提升客户端-服务器通信的并发性能
【6月更文挑战第21天】Java网络编程中,Socket结合多线程提升并发性能,服务器对每个客户端连接启动新线程处理,如示例所示,实现每个客户端的独立操作。多线程利用多核处理器能力,避免串行等待,提升响应速度。防止死锁需减少共享资源,统一锁定顺序,使用超时和重试策略。使用synchronized、ReentrantLock等维持数据一致性。多线程带来性能提升的同时,也伴随复杂性和挑战。
|
22天前
|
安全 Java 网络安全
Java Socket编程教程:构建安全可靠的客户端-服务器通信
【6月更文挑战第21天】构建安全的Java Socket通信涉及SSL/TLS加密、异常处理和重连策略。示例中,`SecureServer`使用SSLServerSocketFactory创建加密连接,而`ReliableClient`展示异常捕获与自动重连。理解安全意识,如防数据截获和中间人攻击,是首要步骤。通过良好的编程实践,确保网络应用在复杂环境中稳定且安全。
|
9天前
|
网络协议 网络架构
【网络编程入门】TCP与UDP通信实战:从零构建服务器与客户端对话(附简易源码,新手友好!)
在了解他们之前我们首先要知道网络模型,它分为两种,一种是OSI,一种是TCP/IP,当然他们的模型图是不同的,如下
|
14小时前
|
安全 网络协议 网络安全
SSL(Secure Sockets Layer)是一种安全协议,用于在客户端和服务器之间建立加密的通信通道。
SSL(Secure Sockets Layer)是一种安全协议,用于在客户端和服务器之间建立加密的通信通道。
4 0
|
3天前
|
Java 数据格式
Java面试题:简述Java Socket编程的基本流程,包括客户端和服务器的创建与通信。
Java面试题:简述Java Socket编程的基本流程,包括客户端和服务器的创建与通信。
11 0
|
13天前
|
消息中间件 Serverless 网络性能优化
消息队列 MQ产品使用合集之客户端和服务器之间的保活心跳检测间隔是怎么设置的
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
|
22天前
|
网络协议 Java Linux
探索Java Socket编程:实现跨平台客户端-服务器通信的奥秘
【6月更文挑战第21天】Java Socket编程示例展示了如何构建跨平台聊天应用。服务器端使用`ServerSocket`监听客户端连接,每个连接启动新线程处理。客户端连接服务器,发送并接收消息。Java的跨平台能力确保代码在不同操作系统上无需修改即可运行,简化开发与维护。
|
3天前
|
存储 关系型数据库 数据库
给阿里云的建议和意见 一个云服务器架构是否可行
摘要(Markdown格式): 在修复阿里云服务器IPv4设置错误时遇到困难,导致服务器远程登录失败及外网访问受阻,耗时三天解决。建议阿里云更新文档,确保设置指导与实际情况一致,例如只需在路由表添加条目关联IPv4。此外,建议优化帮助页面,如采用折叠式设计减少干扰。服务器主要任务是数据分析、存储和分发,文中提出简化服务器框架,消除硬件软件复杂配置,利于初学者和独立开发者快速上手,降低时间成本。该设计旨在减少无用组件,节省资源,同时降低云服务商的人力和支持成本。期望云服务商考虑此类架构创新。目前未知是否有类似产品,期待业界反馈。
206 0
给阿里云的建议和意见 一个云服务器架构是否可行
|
6天前
|
存储 编解码 网络协议
阿里云服务器计算型和通用型四代云服务器实例区别及选择参考
目前阿里云在售的云服务器中,计算型和通用型实例规格都包含了第5代、第6代、第7代和最新第八代倚天云服务器产品,例如计算型实例中有c5、c6、c7、c8y实例,而通用型实例有g5、g6、g7、g8y等实例,有的新手用户并不清楚这四代产品之间的差别,本文为大家展示这四代云服务器实例在规格、CPU(核)、内存(G)、计算、存储、内存以及不同配置的指标数据等方面为大家做个对比,让大家了解一下他们之间的不同,以供参考和选择。
阿里云服务器计算型和通用型四代云服务器实例区别及选择参考