一、消息中间件的作用

(一) 应用解耦

比如在登录中，服务端收到注册消息后把用户信息写入数据库中，再调用推送系统向用户发送短信告知注册成功。然而，推送系统有可能跟这个应用是分开的，我们可以通过消息中间件将该消息传递到推送系统，推送系统收到消息后，调用service向用户推送短信。
有同学说了，那我使用dubbo也能完成相应的功能呀。是的，但是没有消息中间件的快，因为它是一个异步的，服务端只要把消息发出去就可以响应用户了而不用等待推送系统返回结果

(二) 流量削峰

在秒杀系统中，用户发起的秒杀请求首先会到达消息中间件中，再由秒杀系统向消息队列中按照顺序取出消息，剩下的消息退回，告诉用户秒杀失败。

二、ActiveMq基本组件

(一) ConnectionFactory

该类是用来创建connection连接的

//定义链接工厂
ConnectionFactory connectionFactory = null;
connectionFactory = new ActiveMQConnectionFactory("admin", "admin", "tcp://192.168.70.151:61616");

(二) Connection：客户端跟中间件的连接

//创建连接对象
connection = connectionFactory.createConnection();
			
//启动连接
connection.start();

(三) Session：客户端跟中间件的会话

/**
* transacted:是否使用事务 可选值为：true|false
 *            true:使用事务 当设置次变量值。Session.SESSION_TRANSACTED
 *            false:不适用事务,设置次变量 则acknowledgeMode参数必须设置
 * acknowledgeMode:
* Session.AUTO_ACKNOWLEDGE:自动消息确认机制
 * Session.CLIENT_ACKNOWLEDGE:客户端确认机制
* Session.DUPS_OK_ACKNOWLEDGE:有副本的客户端确认消息机制
 */
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

(四) destination

点对点模式：queue
订阅模式：topic

//创建目的地，目的地名称即队列的名称。消息的消费者需要通过此名称访问对应的队列
destination = session.createQueue("my-destination");

destination = session.createTopic("my-destination");

(五) Producer && Consumer

//创建消息的生产者 
MessageProducer producer = session.createProducer(destination);

//创建消息的消费者
MessageConsumer consumer = session.createConsumer(destination);

(六) Message

//创建文本消息
message = session.createTextMessage(”hello“);
producer.send(message);

//接收文本消息
TextMessage message = (TextMessage)consumer.receive();
string text = message.getText();
//创建对象消息
message = session.createObjectMessage(obj);
producer.send(message);
//接收对象消息
ObjectMessage message = (ObjectMessage)consumer.receive();
User user = message.getObject();

三、Springboot集成activeMq

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

(一) yml配置

spring:
 activemq:
  broker-url: tcp://127.0.0.1:61616
  user: admin
  password: admin
  jms:
   pub-sub-domain: true # 默认为false:queue   true:topic
queue: queue_mq # 点对点消费名字
topic: topic_mq # 订阅式消费名字

(二) destination配置

import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jms.annotation.EnableJms;

import javax.jms.Queue;
import javax.jms.Topic;

@Configuration
@EnableJms
public class ActiveMqConfig {

    @Value("${queue}")//对应yml文件中定义的queue
    private String queue;

    @Value("${topic}")//对应yml文件中定义的topic
    private String topic;
    /**
     * 创建点对点的队列  一个消息只能被一个消费者消费  --- 一对一
     * @return
     */
    @Bean
    public Queue queue(){
        return new ActiveMQQueue(queue);
    }
    /**
     * 创建订阅式的队列  一个消息可以被多个消费者消费 --- 一对多
     * @return
     */
    @Bean
    public Topic topic(){
        return new ActiveMQTopic(topic);
    }
}

(三) producer

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import javax.jms.Queue;
import javax.jms.Topic;

@RestController
public class ProducerController {

    @Autowired
    private Queue queue;

    @Autowired
    private Topic topic;

    @Autowired
    private JmsMessagingTemplate jmsMessagingTemplate;

    /**
     * 点对点的消息队列的生产者
     * @param string
     */
    @GetMapping("/queue")
    public void sendMsgQueue(@RequestParam String string){
        System.out.println("消息已经发送,准备被消费,消息为 ---> "+string);
        jmsMessagingTemplate.convertAndSend(queue,string);
    }

    /**
     * 一对多的消息队列的生产者
     * @param string
     */
    @GetMapping("/topic")
    public void sendMsgTopic(@RequestParam String string){
        System.out.println("消息已经发送,准备被消费,消息为 ---> "+string);
        jmsMessagingTemplate.convertAndSend(topic,string);
    }

}

(四) consumer

import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;

@Component
public class TopicConsumer {
    
    /**
     * 监听消息,名字为生产者发送的名字,要一致,不然监听不到.
     * 因为是订阅者模式,可以有多个消费者,我们这里举两个来进行测试
     * @param string
     */
    @JmsListener(destination = "${topic}")
    public void consumerTopicOne(String string){

        System.out.println("我是消费者一号:消费消息成功,信息为---> "+string);

    }

    @JmsListener(destination = "${topic}")
    public void consumerTopicTwo(String string){

        System.out.println("我是消费者二号:消费消息成功,信息为---> "+string);

    }
}