消息

对于消息的生产者与消费者的工作模式，还可以将消息划分成两种模式，同步消费与异步消息。

同步消息就是生产者发送完消息，等待消费者处理，消费者处理完将结果告知生产者，然后生产者继续向下执行业务。这种模式过于卡生产者的业务执行连续性，在现在的企业级开发中，上述这种业务场景通常不会采用消息的形式进行处理。

异步消息就是生产者发送完消息，无需等待消费者处理完毕，生产者继续向下执行其他动作。比如生产者发送了一个日志信息给日志系统，发送过去以后生产者就向下做其他事情了，无需关注日志系统的执行结果。日志系统根据接收到的日志信息继续进行业务执行，是单纯的记录日志，还是记录日志并报警，这些和生产者无关，这样生产者的业务执行效率就会大幅度提升。并且可以通过添加多个消费者来处理同一个生产者发送的消息来提高系统的高并发性，改善系统工作效率，提高用户体验。一旦某一个消费者由于各种问题宕机了，也不会对业务产生影响，提高了系统的高可用性。

Java处理消息的标准规范

目前企业级开发中广泛使用的消息处理技术共三大类，具体如下：

JMS

JMS（Java Message Service）： 这是一个规范，作用等同于JDBC规范，提供了与消息服务相关的API接口。

JMS规范中规范了消息有两种模型。分别是点对点模型和发布订阅模型。

• 点对点模型：peer-2-peer，生产者会将消息发送到一个保存消息的容器中，通常使用队列模型，使用队列保存消息。一个队列的消息只能被一个消费者消费，或未被及时消费导致超时。这种模型下，生产者和消费者是一对一绑定的。
• 发布订阅模型：publish-subscribe，生产者将消息发送到一个保存消息的容器中，也是使用队列模型来保存。但是消息可以被多个消费者消费，生产者和消费者完全独立，相互不需要感知对方的存在。

以上分类是从消息的生产和消费过程来进行区分，针对消息所包含的信息不同，还可以进行不同类别的划分。

JMS消息种类： 根据消息中包含的数据种类划分，可以将消息划分成6种消息。
TextMessage，MapMessage，BytesMessage ，StreamMessage，ObjectMessage，Message （只有消息头和属性）

JMS实现： ActiveMQ、Redis、HornetMQ (RabbitMQ、RocketMQ没有完全满足其规范)

JMS在使用过程中开始被人诟病，原因在于JMS的设计是J2EE规范，站在Java开发的角度思考问题。但是现实往往是复杂度很高的。比如有一个.NET开发的系统A，有一个Java开发的系统B，现在要从A系统给B系统发业务消息，结果两边数据格式不统一，没法操作。JMS不是可以统一数据格式吗？提供了6种数据种类，总有一款适合的。NO，一个都不能用。因为A系统的底层语言不是Java语言开发的，根本不支持那些对象。这就意味着如果想使用现有的业务系统A继续开发已经不可能了，必须推翻重新做使用Java语言开发的A系统。

AMQP

AMQP（advanced message queuing protocol）： 一种协议（高级消息队列协议，也是消息代理规范），规范了网络交换的数据格式，兼容JMS
优点： 具有跨平台性，服务器供应商，生产者，消费者可以使用不同的语言来实现
AMQP消息种类： byte[]
AMQP消息模型： direct exchange，fanout exchange，topic exchange，headers exchange，system exchange
AMQP实现： RabbitMQ、StormMQ、RocketMQ

MQTT

MQTT（Message Queueing Telemetry Transport）： 消息队列遥测传输，专为小设备设计，是物联网（IOT）生态系统中主要成分之一。由于与JavaEE企业级开发没有交集，此处不作过多的说明。

kafka

Kafka，一种高吞吐量的分布式发布订阅消息系统，提供实时消息功能。Kafka技术并不是作为消息中间件为主要功能的产品，但是其拥有发布订阅的工作模式，也可以充当消息中间件来使用，而且目前企业级开发中其身影也不少见。

案例准备

手机验证码案例需求如下：

• 执行下单业务时（模拟此过程），调用消息服务，将要发送短信的订单id传递给消息中间件
• 消息处理服务接收到要发送的订单id后输出订单id（模拟发短信）

由于不涉及数据读写，仅开发业务层与表现层，其中短信处理的业务代码独立开发，代码如下：

订单业务层接口：

public class OrderServiceImpl implements OrderService {@Autowiredprivate MessageService messageService;@Overridepublic void order(String id) {//一系列操作，包含各种服务调用，处理各种业务System.out.println("订单处理开始");//短信消息处理messageService.sendMessage(id);System.out.println("订单处理结束");System.out.println();}
}

订单表现层

@RestController
@RequestMapping("/orders")
public class OrderController {@Autowiredprivate OrderService orderService;@PostMapping("{id}")public void order(@PathVariable String id){orderService.order(id);}
}

短信处理业务层接口

@Service
public class MessageServiceImpl implements MessageService {private ArrayList<String> msgList = new ArrayList<String>();@Overridepublic void sendMessage(String id) {System.out.println("待发送短信的订单已纳入处理队列，id："+id);msgList.add(id);}@Overridepublic String doMessage() {String id = msgList.remove(0);System.out.println("已完成短信发送业务，id："+id);return id;}
}

表现层接口

@RestController
@RequestMapping("/msgs")
public class MessageController {@Autowiredprivate MessageService messageService;@GetMappingpublic String doMessage(){String id = messageService.doMessage();return id;}
}

整合ActiveMQ

1.导入springboot整合ActiveMQ的starter

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

2.配置ActiveMQ的服务器地址

spring:activemq:broker-url: tcp://localhost:61616

3.使用JmsMessagingTemplate操作ActiveMQ

@Service
public class MessageServiceActivemqImpl implements MessageService {@Autowiredprivate JmsMessagingTemplate messagingTemplate;@Overridepublic void sendMessage(String id) {System.out.println("待发送短信的订单已纳入处理队列，id："+id);messagingTemplate.convertAndSend("order.queue.id",id);}@Overridepublic String doMessage() {String id = messagingTemplate.receiveAndConvert("order.queue.id",String.class);System.out.println("已完成短信发送业务，id："+id);return id;}
}

4.使用消息监听器在服务器启动后，监听指定位置，当消息出现后，立即消费消息

@Component
public class MessageListener {@JmsListener(destination = "order.queue.id")@SendTo("order.other.queue.id")public String receive(String id){System.out.println("已完成短信发送业务，id："+id);return "new:"+id;}
}

5.切换消息模型由点对点模型到发布订阅模型，修改jms配置即可

spring:activemq:broker-url: tcp://localhost:61616jms:pub-sub-domain: true

总结

1. springboot整合ActiveMQ提供了JmsMessagingTemplate对象作为客户端操作消息队列
2. 操作ActiveMQ需要配置ActiveMQ服务器地址，默认端口61616
3. 企业开发时通常使用监听器来处理消息队列中的消息，设置监听器使用注解@JmsListener
4. 配置jms的pub-sub-domain属性可以在点对点模型和发布订阅模型间切换消息模型

整合RabbitMQ

RabbitMQ参考：https://blog.csdn.net/weixin_43994244/article/details/128648348
1.入springboot整合amqp的starter，amqp协议默认实现为rabbitmq方案

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

2.配置RabbitMQ的服务器地址

spring:rabbitmq:host: localhostport: 5672

3.初始化直连模式系统设置， 由于RabbitMQ不同模型要使用不同的交换机，因此需要先初始化RabbitMQ相关的对象，例如队列，交换机等

@Configuration
public class RabbitConfigDirect {@Beanpublic Queue directQueue(){return new Queue("direct_queue");}@Beanpublic Queue directQueue2(){return new Queue("direct_queue2");}@Beanpublic DirectExchange directExchange(){return new DirectExchange("directExchange");}@Beanpublic Binding bindingDirect(){return BindingBuilder.bind(directQueue()).to(directExchange()).with("direct");}@Beanpublic Binding bindingDirect2(){return BindingBuilder.bind(directQueue2()).to(directExchange()).with("direct2");}
}

4.使用AmqpTemplate操作RabbitMQ

@Service
public class MessageServiceRabbitmqDirectImpl implements MessageService {@Autowiredprivate AmqpTemplate amqpTemplate;@Overridepublic void sendMessage(String id) {System.out.println("待发送短信的订单已纳入处理队列（rabbitmq direct），id："+id);amqpTemplate.convertAndSend("directExchange","direct",id);}
}

5.使用消息监听器在服务器启动后，监听指定位置，当消息出现后，立即消费消息

@Component
public class MessageListener {@RabbitListener(queues = "direct_queue")public void receive(String id){System.out.println("已完成短信发送业务(rabbitmq direct)，id："+id);}
}

整合(topic模型)

步骤①：同上，步骤②：同上

步骤③：初始化主题模式系统设置

@Configuration
public class RabbitConfigTopic {@Beanpublic Queue topicQueue(){return new Queue("topic_queue");}@Beanpublic Queue topicQueue2(){return new Queue("topic_queue2");}@Beanpublic TopicExchange topicExchange(){return new TopicExchange("topicExchange");}@Beanpublic Binding bindingTopic(){return BindingBuilder.bind(topicQueue()).to(topicExchange()).with("topic.*.id");}@Beanpublic Binding bindingTopic2(){return BindingBuilder.bind(topicQueue2()).to(topicExchange()).with("topic.orders.*");}

主题模式支持routingKey匹配模式，*表示匹配一个单词，#表示匹配任意内容，这样就可以通过主题交换机将消息分发到不同的队列中，详细内容请参看RabbitMQ系列课程。

4.使用AmqpTemplate操作RabbitMQ

@Service
public class MessageServiceRabbitmqTopicImpl implements MessageService {@Autowiredprivate AmqpTemplate amqpTemplate;@Overridepublic void sendMessage(String id) {System.out.println("待发送短信的订单已纳入处理队列（rabbitmq topic），id："+id);amqpTemplate.convertAndSend("topicExchange","topic.orders.id",id);}
}

发送消息后，根据当前提供的routingKey与绑定交换机时设定的routingKey进行匹配，规则匹配成功消息才会进入到对应的队列中。
5.使用消息监听器在服务器启动后，监听指定队列

@Component
public class MessageListener {@RabbitListener(queues = "topic_queue")public void receive(String id){System.out.println("已完成短信发送业务(rabbitmq topic 1)，id："+id);}@RabbitListener(queues = "topic_queue2")public void receive2(String id){System.out.println("已完成短信发送业务(rabbitmq topic 22222222)，id："+id);}
}

使用注解@RabbitListener定义当前方法监听RabbitMQ中指定名称的消息队列。

总结

1. springboot整合RabbitMQ提供了AmqpTemplate对象作为客户端操作消息队列
2. 操作ActiveMQ需要配置ActiveMQ服务器地址，默认端口5672
3. 企业开发时通常使用监听器来处理消息队列中的消息，设置监听器使用注解@RabbitListener
4. RabbitMQ有5种消息模型，使用的队列相同，但是交换机不同。交换机不同，对应的消息进入的策略也不同

整合RocketMQ

在RocketMQ中，处理业务的服务器称为broker，生产者与消费者不是直接与broker联系的，而是通过命名服务器进行通信。broker启动后会通知命名服务器自己已经上线，这样命名服务器中就保存有所有的broker信息。当生产者与消费者需要连接broker时，通过命名服务器找到对应的处理业务的broker，因此命名服务器在整套结构中起到一个信息中心的作用。并且broker启动前必须保障命名服务器先启动。

1.导入springboot整合RocketMQ的starter，此坐标不由springboot维护版本

<dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-spring-boot-starter</artifactId><version>2.2.1</version>
</dependency>

2.配置RocketMQ的服务器地址

rocketmq:name-server: localhost:9876producer: #生产者分配组group: group_rocketmq

3.使用RocketMQTemplate操作RocketMQ

@Service
public class MessageServiceRocketmqImpl implements MessageService {@Autowiredprivate RocketMQTemplate rocketMQTemplate;@Overridepublic void sendMessage(String id) {System.out.println("待发送短信的订单已纳入处理队列（rocketmq），id："+id);//异步调用结果 SendCallback callback = new SendCallback() {@Overridepublic void onSuccess(SendResult sendResult) {System.out.println("消息发送成功");}@Overridepublic void onException(Throwable e) {System.out.println("消息发送失败！！！！！");}};//发送异步消息rocketMQTemplate.asyncSend("order_id",id,callback);}
}

4.使用消息监听器在服务器启动后，监听指定位置，当消息出现后，立即消费消息

@Component
//指定消息队列，消费者组
@RocketMQMessageListener(topic = "order_id",consumerGroup = "group_rocketmq")
public class MessageListener implements RocketMQListener<String> {//消息类型@Overridepublic void onMessage(String id) {System.out.println("已完成短信发送业务(rocketmq)，id："+id);}
}

RocketMQ的监听器必须按照标准格式开发，实现RocketMQListener接口，泛型为消息类型。
使用注解@RocketMQMessageListener定义当前类监听RabbitMQ中指定组、指定名称的消息队列。

总结

1. springboot整合RocketMQ使用RocketMQTemplate对象作为客户端操作消息队列
2. 操作RocketMQ需要配置RocketMQ服务器地址，默认端口9876
3. 企业开发时通常使用监听器来处理消息队列中的消息，设置监听器使用注解@RocketMQMessageListener

整合Kafka

1.导入springboot整合Kafka的starter，此坐标由springboot维护版本

<dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId>
</dependency>

2.配置Kafka的服务器地址

spring:kafka:bootstrap-servers: localhost:9092consumer:group-id: order

3.使用KafkaTemplate操作Kafka

@Service
public class MessageServiceKafkaImpl implements MessageService {@Autowiredprivate KafkaTemplate<String,String> kafkaTemplate;@Overridepublic void sendMessage(String id) {System.out.println("待发送短信的订单已纳入处理队列（kafka），id："+id);kafkaTemplate.send("itheima2022",id);}
}

4.使用消息监听器在服务器启动后，监听指定位置，当消息出现后，立即消费消息

@Component
public class MessageListener {@KafkaListener(topics = "itheima2022")public void onMessage(ConsumerRecord<String,String> record){System.out.println("已完成短信发送业务(kafka)，id："+record.value());}
}

使用注解@KafkaListener定义当前方法监听Kafka中指定topic的消息，接收到的消息封装在对象ConsumerRecord中，获取数据从ConsumerRecord对象中获取即可。

总结

1. springboot整合Kafka使用KafkaTemplate对象作为客户端操作消息队列
2. 操作Kafka需要配置Kafka服务器地址，默认端口9092
3. 企业开发时通常使用监听器来处理消息队列中的消息，设置监听器使用注解@KafkaListener。接收消息保存在形参ConsumerRecord对象中