springboot--ActiveMQ--消息队列

ActiveMQ远程消息队列

一、我们创建springboot项目工程

二、创建完毕我们简单介绍 activeMQ

1.概述

消息中间件可以理解成就是一个服务软件,保存信息的容器,比如生活中的快递云柜.
我们把数据放到消息中间件当中, 然后通知对应的服务进行获取
消息中间件是在消息的传输过程中保存信息的容器

2.消息中间件应用场景

1. 使用消息服务器当做大的队列使用, 先进先出, 来处理高并发写入操作
2. 使用消息服务器可以将业务系统的串行执行改为并行执行, 处理效率高, 更合理的榨取服务器的性能.

3.同步与异步技术

同步技术
	dubbo是一中同步技术, 实时性高, controller调用service项目, 调用就执行, 
	如果service项目中的代码没有执行完, controller里面的代码一致等待结果.
异步技术
	mq消息中间件技术(jms) 是一种异步技术, 消息发送方, 将消息发送给消息服务器, 
	消息服务器未必立即处理.什么时候去处理, 主要看消息服务器是否繁忙, 
	消息进入服务器后会进入队列中, 先进先出.实时性不高.

三、JMS 介绍

概述:

jms的全称叫做Java message service (Java消息服务) jms是jdk底层定义的规范
各大厂商都是实现这个规范的技术

jms消息服务器同类型技术

1.ActiveMQ
	是apache的一个比较老牌的消息中间件, 它比较均衡, 既不是最安全的, 也不是最快的.
2.RabbitMQ
	是阿里巴巴的一个消息中间件, 更适合金融类业务, 它对数据的安全性比较高.能够保证数据不丢失.
3.Kafka
	Apache下的一个子项目。特点：高吞吐，在一台普通的服务器上既可以达到10W/s的吞吐速率；适合处理海量数据。

JMS中支持的消息类型:

TextMessage
	一个字符串对象
MapMessage
	key-value
ObjectMessage
	一个序列化的 Java 对象
BytesMessage
	一个字节的数据流
StreamMessage
	Java 原始值的数据流

JMS中的两种发送模式

1.点对点模式
	一个发送方, 一个接收方. 也可以多个发送方, 一个接收方, 主要是接收方必须是第一个.
	示例图
		
2.订阅发布模式
	一个发送方, 多个接收方.	发送方也可以是多个, 主要看接收方, 接收方必须是多个
	示例图

ActiveMQ服务器安装

官方网站: http:activemq.apache.org/

四、项目创建完毕后引入依赖

 
      
   
            org.springframework.boot
            spring-boot-starter-activemq
        
        
        
            org.apache.activemq
            activemq-pool
            5.15.0
        
       
            org.springframework.boot
            spring-boot-starter-web
        

写入yml配置

server:
  port: 8080
spring:
  activemq:
    broker-url: tcp://www.yangbuyi.top:61616 # 可以使用我的服务器mq
    user: admin
    password: admin
    close-timeout: 15s   # 在考虑结束之前等待的时间
    in-memory: true      # 默认代理URL是否应该在内存中。如果指定了显式代理，则忽略此值。
    non-blocking-redelivery: false  # 是否在回滚回滚消息之前停止消息传递。这意味着当启用此命令时，消息顺序不会被保留。
    send-timeout: 0     # 等待消息发送响应的时间。设置为0等待永远。
    queue-name: active.queue  # 点对点服务名称  可修改
    topic-name: active.topic.name.model # 订阅服务名称 可修改
    packages:
      trust-all: true   # 配置信任 反序列化对象要用到的

#  packages:
#    trust-all: true #不配置此项，会报错
pool:
  enabled: true
  max-connections: 10   #连接池最大连接数
  idle-timeout: 30000   #空闲的连接过期时间，默认为30秒

消息配置类创建

package top.yangbuyi.config;

import org.apache.activemq.ActiveMQConnectionFactory;
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.config.JmsListenerContainerFactory;
import org.springframework.jms.config.SimpleJmsListenerContainerFactory;
import org.springframework.jms.core.JmsMessagingTemplate;

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

/**
 * @description: 杨不易网站:www.yangbuyi.top
 * @program: yangbuyi_mq
 * @ClassName: BeanConfig
 * @create: 2020-07-30 09:18
 * @author: yangbuyi
 * @since： JDK1.8
 * @BeanConfig:
 **/

@Configuration
public class BeanConfig {

	  @Value("${spring.activemq.broker-url}")
	  private String brokerUrl;

	  @Value("${spring.activemq.user}")
	  private String username;

	  @Value("${spring.activemq.topic-name}")
	  private String password;

	  @Value("${spring.activemq.queue-name}")
	  private String queueName;

	  @Value("${spring.activemq.topic-name}")
	  private String topicName;

	  @Bean(name = "queue")
	  public Queue queue() {
			return new ActiveMQQueue(queueName);
	  }

	  @Bean(name = "topic")
	  public Topic topic() {
			return new ActiveMQTopic(topicName);
	  }

	  /**
	  * @Description: 杨不易个人网址:http://yangbuyi.top
	  * 功能描述:	 初始化工厂
	  * @Param:
	  * @return:
	  * @Author: TeouBle
	  * @Date: 2020/7/30 10:24
	  */
	  @Bean
	  public ConnectionFactory connectionFactory() {
			ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(username, password, brokerUrl);
			/* 配置信任 反序列化 */
			activeMQConnectionFactory.setTrustAllPackages(true);
			return activeMQConnectionFactory;
	  }

	  /**
	  * @Description: 杨不易个人网址:http://yangbuyi.top
	  * 功能描述: 	JMS 模板 Java消息服务
	  * @Param:
	  * @return:
	  * @Author: TeouBle
	  * @Date: 2020/7/30 10:25
	  */
	  @Bean
	  public JmsMessagingTemplate jmsMessageTemplate() {
			return new JmsMessagingTemplate(connectionFactory());
	  }

	  /**
	  * @Description: 杨不易个人网址:http://yangbuyi.top
	  * 功能描述:	在Queue模式中，对消息的监听需要对containerFactory进行配置
	  * @Param: 
	  * @return: 
	  * @Author: TeouBle
	  * @Date: 2020/7/30 10:25
	  */
	  @Bean("queueListener")
	  public JmsListenerContainerFactory queueJmsListenerContainerFactory(ConnectionFactory connectionFactory) {
			SimpleJmsListenerContainerFactory factory = new SimpleJmsListenerContainerFactory();
			factory.setConnectionFactory(connectionFactory);
			factory.setPubSubDomain(false);
			return factory;
	  }

	  /**
	  * @Description: 杨不易个人网址:http://yangbuyi.top
	  * 功能描述:在Topic模式中，对消息的监听需要对containerFactory进行配置
	  * @Param: 
	  * @return: 
	  * @Author: TeouBle
	  * @Date: 2020/7/30 10:25
	  */
	  @Bean("topicListener")
	  public JmsListenerContainerFactory topicJmsListenerContainerFactory(ConnectionFactory connectionFactory) {
			SimpleJmsListenerContainerFactory factory = new SimpleJmsListenerContainerFactory();
			factory.setConnectionFactory(connectionFactory);
			factory.setPubSubDomain(true);
			return factory;
	  }
}

创建消息监听类

QueueConsumerListener 点对点模式 消费者 接收

package top.yangbuyi.listener;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @description: 杨不易网站:www.yangbuyi.top
 * @program: yangbuyi_mq
 * @ClassName: QueueConsumerListener
 * @create: 2020-07-30 09:21
 * @author: yangbuyi
 * @since： JDK1.8
 * @QueueConsumerListener:
 **/
@Component
@Slf4j
public class QueueConsumerListener {
	  @Value("${spring.activemq.queue-name}")
	  private String queueName;

	  //queue模式的消费者
	  @JmsListener(destination = "${spring.activemq.queue-name}", containerFactory = "queueListener")
	  public void readActiveQueue(String message) {
			System.out.println("queue接受到：" + message);
			System.out.println("点对点服务名称:"+ queueName);
			System.out.println("点对点监听成功");
	  }

}

TopicConsumerListener 订阅模式 消费者 接收

package top.yangbuyi.listener;

import lombok.extern.slf4j.Slf4j;
import org.apache.catalina.User;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.configurationprocessor.json.JSONObject;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import top.yangbuyi.controller.user;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.ObjectMessage;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;

/**
 * @description: 杨不易网站:www.yangbuyi.top
 * @program: yangbuyi_mq
 * @ClassName: TopicConsumerListener
 * @create: 2020-07-30 09:23
 * @author: yangbuyi
 * @since： JDK1.8
 * @TopicConsumerListener:
 **/

@Component
public class TopicConsumerListener {
	  @Value("${spring.activemq.topic-name}")
	  private String topicName;

	  //topic模式的消费者
	  @JmsListener(destination = "${spring.activemq.topic-name}", containerFactory = "topicListener")
	  public void readActiveQueue(Message message) {
			ObjectMessage u = (ObjectMessage) message;
			try {
				  user object = (user) u.getObject();
				  System.out.println(object.getName());
				  System.out.println(object.getAge());
			} catch (JMSException e) {
				  e.printStackTrace();
			}
			System.out.println();
			System.out.println("topic接受到：" + message);
			System.out.println("订阅服务名称:" + topicName);
	  }
}

创建一个小对象用来测试传递对象

package top.yangbuyi.controller;


import java.io.Serializable;

/**
 * @description: 杨不易网站:www.yangbuyi.top
 * @program: yangbuyi_mq
 * @ClassName: user
 * @create: 2020-07-30 10:03
 * @author: yangbuyi
 * @since： JDK1.8
 * @user:
 **/


public class user implements Serializable {
	private String name;
	private Integer age;

	  public user(String name, Integer age) {
			this.name = name;
			this.age = age;
	  }

	  public String getName() {
			return name;
	  }

	  public void setName(String name) {
			this.name = name;
	  }

	  public Integer getAge() {
			return age;
	  }

	  public void setAge(Integer age) {
			this.age = age;
	  }
}

创建controller

package top.yangbuyi.controller;

import javafx.geometry.Pos;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.MediaType;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.web.bind.annotation.*;

import javax.jms.Destination;
import javax.jms.Queue;
import javax.jms.Topic;
import java.util.ArrayList;

/**
 * @description: 杨不易网站:www.yangbuyi.top
 * @program: yangbuyi_mq
 * @ClassName: ProducerController
 * @create: 2020-07-30 09:20
 * @author: yangbuyi
 * @since： JDK1.8
 * @ProducerController:  生产者
 **/

@RestController
@Slf4j
public class ProducerController {
	  @Autowired
	  private JmsMessagingTemplate jmsMessagingTemplate;

	  @Autowired
	  private Queue queue;

	  @Autowired
	  private Topic topic;

	  @RequestMapping(value ="queue/test",method = RequestMethod.POST)
	  public String sendQueue(@RequestBody String str) {
			this.sendMessage(this.queue, str);
			return "success---点对点发送成功";
	  }

	  @RequestMapping(value = "topic/test",method = RequestMethod.POST)
	  public String sendTopic(@RequestBody String str) {
			this.sendMessageObject(this.topic, new user(str,10));
			return "success---订阅发送成功";
	  }


	  // 发送消息，destination是发送到的队列，message是待发送的消息
	  // 生产者
	  private void sendMessageObject(Destination destination, final Object message){
			jmsMessagingTemplate.convertAndSend(destination, message);

	  }

	  // 发送消息，destination是发送到的队列，message是待发送的消息
	  // 生产者
	  private void sendMessage(Destination destination, final String message){
			jmsMessagingTemplate.convertAndSend(destination, message);
	  }


}

五、进行postman发送请求

订阅服务

点对点服务

到此activemq就讲到这里了 还有蛮多可以自己去 学习学习!!!

你的压力来源于无法自律，只是假装努力，现状跟不上内心欲望，所以你焦虑又恐慌。——杨不易