JMS消息队列（ActiveMQ消息中间件）

深碍√TFBOYSˉ_ 2024-02-19 15:26 17阅读 0赞

作用  
1，减少并发量  
2，异步并解耦（解决系统模块的异步调用）  
3，通过消息收发服务，从一个JMS客户机向另一个JML客户机发送消息  
消息类型  
TextMessage 一个字符串对象（常用）即String  
MapMessage 键值对对象（常用）  
ObjectMessage 序列化java对象  
BytesMessage 字节数据流  
StreamMesaage 原始值数据流  
消息传递类型（Destination 父接口）  
queue （子接口）：point to pint 点对点 （ptp）：一个生产者对应一个消费者  
topic （子接口）：publish/subscribe 发布/订阅 ：一对多，一个消费者发消息，多个消费接受

ActiveMQ 下载安装  
1，官方网站下载： [http://activemq.apache.org/][http_activemq.apache.org]

2， 将 apache-activemq-5.12.0-bin.tar.gz 上传至服务器，并解压  
tar zxvf apache-activemq-5.12.0-bin.tar.gz

3，为 apache-activemq-5.12.0 目录赋权  
chmod 777 apache-activemq-5.12.0

4，进入 apache-activemq-5.12.0\\bin 目录，赋与执行权限  
chmod 755 activemq

5，启动 ./activemq start

6，访问ActiveMQ管理页面 [http://192.168.25.135:8161/][http_192.168.25.135_8161] （admin） 账号密码 默认admin

7，连接端口 tcp://192.168.25.135:61616 (http协议是基于tcp协议建立的一个应用)

Spring整合JMS （SpringJMS）  
创建工程 POM 文件中引入 SpringJms 、 activeMQ

<dependency>
        <groupId>org.springframework</groupId>
        <artifactId>spring-jms</artifactId>
        <version>${spring.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.activemq</groupId>
        <artifactId>activemq-client</artifactId>
        <version>5.13.4</version>
    </dependency>

配置文件XML  
1,applicationContext-jms-producer.xml （生产者） file:///E:/JAVA学习资料/项目一\_品优购/品优购资源V1.3/解压文件/配置文件/JMS

<?xml version="1.0" encoding="UTF-8"?>

<context:component-scan base-package="cn.itcast.demo"></context:component-scan>     
    
      
    <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">  
        <property name="brokerURL" value="tcp://192.168.25.129:61616"/>  
    </bean>
       
      
    <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">  
      
        <property name="targetConnectionFactory" ref="targetConnectionFactory"/>  
    </bean>  
           
      
    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">  
          
        <property name="connectionFactory" ref="connectionFactory"/>  
    </bean>    
    
      
    <bean id="queueTextDestination" class="org.apache.activemq.command.ActiveMQQueue">  
        <constructor-arg value="queue_text"/>  
    </bean>    
    
      
    <bean id="topicTextDestination" class="org.apache.activemq.command.ActiveMQTopic">  
        <constructor-arg value="topic_text"/>  
    </bean>

2, applicationContext-jms-consumer-queue.xml (消费者) file:///E:/JAVA学习资料/项目一\_品优购/品优购资源V1.3/解压文件/配置文件/JMS

<?xml version="1.0" encoding="UTF-8"?>

<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">  
        <property name="brokerURL" value="tcp://192.168.25.129:61616"/>  
    </bean>
       
      
    <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">  
      
        <property name="targetConnectionFactory" ref="targetConnectionFactory"/>  
    </bean>  
    
      
    <bean id="queueTextDestination" class="org.apache.activemq.command.ActiveMQQueue">  
        <constructor-arg value="queue_text"/>  
    </bean>    
    
    
    <bean id="myMessageListener" class="cn.itcast.demo.MyMessageListener"></bean>
    
    <bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="destination" ref="queueTextDestination" />
        <property name="messageListener" ref="myMessageListener" />
    </bean>

实体类中的使用： 首先需要注入到使用的类中

@Autowired
    private JmsTemplate jmsTemplate;
    @Autowired
    private Destination smsDestination;
    类中调用（例子）
    jmsTemplate.send(smsDestination, new MessageCreator() {
    @Override
    public Message createMessage(Session session) throws JMSException {
        MapMessage mapMessage = session.createMapMessage();
       // 添加一些信息给消息体
       // mapMessage.setString("mobile", phone);//手机号
       // mapMessage.setString("template_code", "SMS_85735065");//模板编号
       // mapMessage.setString("sign_name", "xxx");//签名
       // Map m=new HashMap<>();
       // m.put("number", code); //验证码写入
        mapMessage.setString("param", JSON.toJSONString(m));//参数
        return mapMessage;
        }
    });

消费者需要实现消息监听类：  
例如：实现MessageListenner 在onmessage方法中调用需要执行的相关方法

@Component
    public class ItemDeleteListener implements MessageListener {
    @Autowired
    private ItemSearchService itemSearchService;
    @Override
    public void onMessage(Message message) {
    ObjectMessage objectMessage = (ObjectMessage) message;
        try {
            Long[] goodsId = (Long[]) objectMessage.getObject();
            itemSearchService.deleteByGoodIds(Arrays.asList(goodsId));
        } catch (JMSException e) {
            e.printStackTrace();
        }
      }
    }

如果是springBoot微服务实现消费者 相对更加简单 下面我们来举例子：  
创建的时候添加一些使用的依赖  
在application.properties 文件中配置：  
server.port=8088  
spring.activemq.broker-url=tcp://192.168.25.135:61616

/**
    * 消息生产者
    */
    @RestController
    public class QueueController {
    @Autowired
    private JmsMessagingTemplate jmsMessagingTemplate;
    @RequestMapping("/send")
    public void send(String text){
            jmsMessagingTemplate.convertAndSend("sms", text);
        }
    }

生产者发送的是什么类型的消息，那么消费者接受的消息类型与之对应

@Component
    public class Consumer {
    @JmsListener(destination="sms")
    public void readMessage(String text){
    System.out.println("接收到消息");
    String mobile = map.get("mobile");
    System.out.println(mobile);
    // 获取开发者账号信息
    //AccountInfo.execute();
    // 验证码通知短信接口
    //IndustrySMS.execute();
    }
    }

如果生产者是 springMVC的配置xml配置文件

<bean id="smsDestination" class="org.apache.activemq.command.ActiveMQQueue">
    <constructor-arg value="sms"/> 
    </bean>

其中sms 对应的是 消息消费者的 @JmsListener(destination=“sms”) 这个注解  
通常使用map 发送消息的情况最多：  
那么接受消息的格式为 ：Map<String,String>

[http_activemq.apache.org]: http://activemq.apache.org/
[http_192.168.25.135_8161]: http://192.168.25.135:8161/