淘淘商城——添加商品同步到索引库以及消息机制测试

我们在添加商品时需要与索引库进行同步，这样每添加一个商品，索引库就多一个文档，这样做的好处是不用把数据库中的所有数据进行同步，大大提高了性能并且节约了时间。
我们要做的是当添加商品的时候发送ActiveMQ消息，至于发送什么类型的ActiveMQ消息则要根据实际应用场景来定，由于在淘淘商城这个项目中添加商品涉及到同步缓存、同步索引库、添加静态页面等操作，也就是一个消息被多个消费者所消费，显然发送topic消息更为合适。
要发送topic消息，就要在taotao-manager-service工程下的applicationContext-activemq.xml配置文件中做下配置，我们将原来配置的topic做下简单的修改，根据命名有意义原则，我们给消息起名为”item-add-topic”，如下图所示。

下一步是我们找到添加商品的ItemServiceImpl实现类，在该类中注入JmsTemplate和Topic，如下图所示。

然后我们找到添加商品的方法，在添加完商品后，发送消息，这里需要考虑一个问题，那就是消息的内容应该是什么？既然是添加商品，消费者肯定是要知道添加的商品是哪个商品，同时本着简单的原则，我们只需要传送新增商品的ID即可，如下图所示。

为了方便大家复制，现把ItemServiceImpl实现类的代码给出。

/**
 * 商品管理Service
 * <p>Title: ItemServiceImpl</p>
 * <p>Description: </p>
 * <p>Company: www.itcast.cn</p> 
 * @version 1.0
 */
@Service
public class ItemServiceImpl implements ItemService {

    @Autowired
    private TbItemMapper itemMapper;

    @Autowired
    private TbItemDescMapper itemDescMapper;

    @Autowired
    private JmsTemplate jmsTemplate;

    @Resource(name="topicDestination")
    private Topic topicDestination;

    @Override
    public TbItem getItemById(long itemId) {
        TbItem item = itemMapper.selectByPrimaryKey(itemId);
        return item;
    }

    @Override
    public EasyUIDataGridResult getItemList(int page, int rows) {
        // 分页处理
        PageHelper.startPage(page, rows);
        // 执行查询
        TbItemExample example = new TbItemExample();
        List<TbItem> list = itemMapper.selectByExample(example);
        // 创建返回结果对象
        EasyUIDataGridResult result = new EasyUIDataGridResult();
        result.setRows(list);
        // 取总记录数
        PageInfo<TbItem> pageInfo = new PageInfo<TbItem>(list);
        result.setTotal(pageInfo.getTotal());
        return result;
    }

    @Override
    public TaotaoResult addItem(TbItem item, String desc) {
        // 先生成商品id
        final long itemId = IDUtils.genItemId();
        item.setId(itemId);
        // 商品状态，1-正常，2-下架，3-删除
        item.setStatus((byte) 1);
        Date date = new Date();
        item.setCreated(date);
        item.setUpdated(date);
        // 插入到商品表
        itemMapper.insert(item);
        // 商品描述
        TbItemDesc itemDesc = new TbItemDesc();
        itemDesc.setItemId(itemId);
        itemDesc.setItemDesc(desc);
        itemDesc.setCreated(date);
        itemDesc.setUpdated(date);
        // 插入商品描述
        itemDescMapper.insert(itemDesc);

        // 商品添加完成后发送一个MQ消息
        jmsTemplate.send(topicDestination, new MessageCreator() {

            @Override
            public Message createMessage(Session session) throws JMSException {
                // 创建一个消息对象
                // 要在匿名内部类访问局部变量itemId，itemId需要用final修饰
                TextMessage message = session.createTextMessage(itemId + "");
                return message;
            }
        });

        return TaotaoResult.ok();
    }

}

上面是发送ActiveMQ消息，下面我们再来编写接收消息的代码，接收消息是在taotao-search-service工程中，我们首先要做的事情就是修改ItemMapper.xml文件，添加一个根据商品ID来查询商品详情的方法，如下图所示。

为了方便大家复制，现把ItemMapper.xml文件的内容给出。

<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
<mapper namespace="com.taotao.search.mapper.ItemMapper">

    <select id="getItemList" resultType="com.taotao.common.pojo.SearchItem">
        SELECT
            a.id,
            a.title,
            a.sell_point,
            a.price,
            a.image,
            b.`name` category_name,
            c.item_desc
        FROM
            tb_item a
        LEFT JOIN tb_item_cat b ON a.cid = b.id
        LEFT JOIN tb_item_desc c ON a.id = c.item_id
        WHERE a.`status`=1
    </select>

    <select id="getItemById" parameterType="long" resultType="com.taotao.common.pojo.SearchItem">
        SELECT
            a.id,
            a.title,
            a.sell_point,
            a.price,
            a.image,
            b.`name` category_name,
            c.item_desc
        FROM
            tb_item a
        LEFT JOIN tb_item_cat b ON a.cid = b.id
        LEFT JOIN tb_item_desc c ON a.id = c.item_id
        WHERE a.`status`=1
            AND a.id=#{id}
    </select>

</mapper>

接着，在ItemMapper接口中添加根据商品ID查询商品详情的方法，如下图所示。

现在我启动的是集群版的Solr服务器，因此要保证applicationContext-solr.xml配置文件当前切换到Solr集群版，如下图所示。

下面我们需要创建一个ItemAddListener监听器类，在该类中处理同步索引库的逻辑，如下图所示。

为了方便大家复制，现把ItemAddListener监听器类的代码给出。

public class ItemAddListener implements MessageListener {

    @Autowired
    private ItemMapper itemMapper;

    @Autowired
    private SolrServer solrServer;

    @Override
    public void onMessage(Message message) {
        // 从消息中取商品id
        try {
            TextMessage textMessage = (TextMessage) message;
            String strItemId = textMessage.getText();
            // 转换成Long
            Long itemId = new Long(strItemId);
            // 根据商品id来查询商品信息
            SearchItem searchItem = itemMapper.getItemById(itemId);
            // 把商品信息添加到索引库
            SolrInputDocument document = new SolrInputDocument();
            // 为文档添加域
            document.addField("id", searchItem.getId());
            document.addField("item_title", searchItem.getTitle());
            document.addField("item_sell_point", searchItem.getSell_point());
            document.addField("item_price", searchItem.getPrice());
            document.addField("item_image", searchItem.getImage());
            document.addField("item_category_name", searchItem.getCategory_name());
            document.addField("item_desc", searchItem.getItem_desc());
            // 向索引库中添加文档。
            solrServer.add(document);
            // 提交
            solrServer.commit();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}

写完监听器我们再配置下接收的topic（要与taotao-manager-service当中配置的topic一致），以及设置监听器，如下图所示。

为了方便大家复制，现把applicationContext-activemq.xml文件的内容给出。

<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:context="http://www.springframework.org/schema/context" xmlns:p="http://www.springframework.org/schema/p"
    xmlns:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.2.xsd
    http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.2.xsd
    http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-4.2.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-4.2.xsd
    http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-4.2.xsd">

    <!-- 真正可以产生Connection的ConnectionFactory，由对应的JMS服务厂商提供 -->
    <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="tcp://192.168.25.129:61616" />
    </bean>
    <!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->
    <bean id="connectionFactory"
        class="org.springframework.jms.connection.SingleConnectionFactory">
        <!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->
        <property name="targetConnectionFactory" ref="targetConnectionFactory" />
    </bean>

    <!--这个是队列目的地，点对点的 -->
    <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg>
            <value>spring-queue</value>
        </constructor-arg>
    </bean>
    <!--这个是主题目的地，一对多的 -->
    <bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">
        <constructor-arg value="item-add-topic" />
    </bean>

    <!-- 配置消息监听器 -->
    <bean id="myMessageListener" class="com.taotao.search.listener.MyMessageListener" />
    <!-- 配置消息监听容器 -->
    <bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="destination" ref="queueDestination" />
        <property name="messageListener" ref="myMessageListener" />
    </bean>

    <!-- 配置消息监听器 -->
    <bean id="itemAddListener" class="com.taotao.search.listener.ItemAddListener" />
    <!-- 配置消息监听容器 -->
    <bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="destination" ref="topicDestination" />
        <property name="messageListener" ref="itemAddListener" />
    </bean>

</beans>

在启动工程前，我们需要把学习全局异常处理时人为添加的异常给注释掉，如下图所示。

好了，代码终于写完了，现在我们来测试一下，我们依次启动taotao-manager、taotao-content、taotao-search、taotao-manager-web、taotao-port-web、taotao-search-web工程。启动成功之后，我们在浏览器地址栏中输入http://localhost:8080来访问淘淘商城后台管理页面，添加一款商品，这里我添加的是锤子手机T2，如下图所示。

添加完商品后，我们到淘淘商城首页进行搜索，看能不能搜索到我们刚才添加的手机，如下图所示，发现可以正常搜索到刚才添加的锤子手机！！说明我们的消息机制没问题。