2019-09-17 ActiveMQ

在项目中，将一些无需即时返回且耗时的操作提取出来，进行了异步处理，
而这种异步处理的方式大大的节省了服务器的请求响应时间，从而提高了系统的
吞吐量。

ActiveMQ 消息的传递有两种类型：
一种是点对点的，即一个生产者和一个消费者一一对应；
一种是发布/订阅模式，即一个生产者产生消息并进行发送后，可以由多个消费
者进行接收。

JMS
JMS 即 Java 消息服务（Java Message Service）应用程序接口，是一个 Java 平
台中关于面向消息中间件（MOM）的 API，用于在两个应用程序之间，或分布式
系统中发送消息，进行异步通信。
•StreamMessage -- Java 原始值的数据流
• MapMessage--一套名称-值对
• TextMessage--一个字符串对象
• ObjectMessage--一个序列化的 Java 对象
• BytesMessage--一个字节的数据流

<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-client</artifactId>
</dependency>
<dependency>
<groupId>javax.jms</groupId>
<artifactId>javax.jms-api</artifactId>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-jms</artifactId>
</dependency>

生产者spring-activemq.xml

    <!--真正可以产生 Connection 的 ConnectionFactory ，由对应的 JMS 服务厂商提供-->

    <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="tcp://192.168.12.168:61616"/>
    </bean>
    <!--Spring 用于管理真正的 ConnectionFactory 的 ConnectionFactory-->
    <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
        <property name="targetConnectionFactory" ref="targetConnectionFactory"/>
    </bean>
    <!--Spring 提供的 JMS 工具类，它可以进行消息发送、接收等-->
    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory" ref="connectionFactory"/>
    </bean>
    <!--队列模式，在商品审核后接收的商品列表队，队列根据需求增加-->
    <bean id="itemEsQueue" class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg value="item_es_queue"/>
    </bean>

生产者控制器

@Autowired
    private JmsTemplate jmsTemplate;
 @Autowired
    private ActiveMQQueue itemEsDeleteQueue;

    private void sendMQMsg(Destination destination, Long[] ids) {
        try {
            jmsTemplate.send(destination, new MessageCreator() {
                @Override
                public Message createMessage(Session session) throws JMSException {
                    ObjectMessage objectMessage = session.createObjectMessage();
                    objectMessage.setObject(ids);
                    return objectMessage;
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }

消费者applicationContext-activemq.xml

<!--真正可以产生 Connection 的 ConnectionFactory ，由对应的 JMS 服务厂商提供-->

    <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="tcp://192.168.12.168:61616"/>
    </bean>
    <!--Spring 用于管理真正的 ConnectionFactory 的 ConnectionFactory-->
    <bean id="jmsConnectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
        <property name="targetConnectionFactory" ref="targetConnectionFactory"/>
    </bean>
    <!--队列模式，在商品审核后接收的商品列表队-->
    <bean id="itemEsQueue" class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg value="pinyougou_item_es_queue"/>
    </bean>
    <!--messageListener 实现类-->
    <bean id="itemImportMessageListener" class="com.pinyougou.search.activemq.listener.ItemImportMessageListener"/>
    <!--配置一个 jms 监听容器-->
    <bean id="jmsImportContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="jmsConnectionFactory"/>
        <property name="destination" ref="itemEsQueue"/>
        <property name="messageListener" ref="itemImportMessageListener"/>
    </bean>

消费者控制器

public class ItemDeleteMessageListener extends AbstractAdaptableMessageListener {
    @Autowired
    private ItemSearchService itemSearchService;
    @Override
    public void onMessage(Message message, Session session) throws JMSException {
        ObjectMessage objectMessage = (ObjectMessage) message;
        Long[] goodsIds = (Long[]) objectMessage.getObject();
        itemSearchService.deleteItemByIds(goodsIds);
        System.out.println("同步删除索引库中数据完成！");
    }
}