如何在分布式环境中同步solr索引库和缓存信息

赵小忠

发布于 2018-01-24 19:39:36

1.3K0

发布于 2018-01-24 19:39:36

文章被收录于专栏：禁心尽力

搜索无处不在，相信各位每天都免不了与它的亲密接触，那么我想你确实有必要来了解一下它们，就上周在公司实现的一个小需求来给各位分享一下：如何在分布式环境下同步索引库？

需求分析

公司数据库中的数据信息每天都免不了增、删、改操作，在执行这些简单的更新操作时，我们不仅将变更后的数据要更新到数据库中，同时还要马上同步索引库中的数据，有的时候还要同步一下缓存中的数据(本文只分享如何同步solr索引库)。

分析方案

当我们在后台管理系统中触发了更新操作时，不会紧跟着调用同步功能去更新索引库和缓存这种机制去实现，因为耦合性太高了，容易影响正常的业务流程。那么，既然我们不做，做的话就要影响业务，所以我们就有必要请一位私人秘书来替我们完成同步操作了，既然请了秘书，就没必要再去关心同步操作，而是我们只需要在更新完数据后通知这位秘书，让它去完成同步操作，岂不更妙？好了，说了这么久，这位秘书就是英俊潇洒不可或缺的消息队列——MQ，为什么使用它？主要还是开源、解耦。废话不说了，一起从简，开始上码。

哦，对了到这儿我就有必要说一下MQ的俩种使用模式，因为这个确实有点用，我就爬过这坑。主要分为2种：点对点(Queue)和发布\订阅(Topic)模式。

从上图可以看出，这俩种模式最主要的区别就是发送出去的消息可以由多少个消费者来接受，很明显：

发布\订阅模式：需要一个生产者发送消息到主题版块（Topic）中，可以有多个消费者订阅该版块来接受消息。消费者接受消息时，必须处于运行状态，而且只能接受运行之后的消息。

点对点模式：需要一个生产者发送消息到队列版块（Queue）中，只能有一个消费者从该队列（Queue）中接受该消息。生产者发送消息时，消费者不需要处于运行状态。

好，明确这点就够了，我们先用起来，至于它的一些细节，你们自己去找找资料好好读读，因为本人也是初次使用到，后期有机会再和大家共勉。

一路走好，码到成功

步骤一：安装MQ（activeMQ）

狠简单，解压即用。如有需要请参考http://www.cnblogs.com/1315925303zxz/p/6377551.html。

步骤二：spring整合MQ

 1 　　 <!-- 真正可以产生Connection的ConnectionFactory，由对应的 JMS服务厂商提供 -->
 2     <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
 3         <property name="brokerURL" value="tcp://192.168.136.139:61616"/>
 4     </bean>
 5     <!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->
 6     <bean id="connectionFactory"
 7         class="org.springframework.jms.connection.SingleConnectionFactory">
 8         <!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->
 9         <property name="targetConnectionFactory" ref="targetConnectionFactory"/>
10     </bean>
11     
12     <!-- 生产者 -->
13     <!-- Spring提供的JMS工具类，它可以进行消息发送、接收等 -->
14     <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
15         <!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->
16         <property name="connectionFactory" ref="connectionFactory"/>
17     </bean>
18     <!--这个是队列目的地：(俩种配置方式)
19             一、点对点模式：需要一个生产者发送消息到队列版块（Queue）中，只能有一个消费者从该队列（Queue）中接受该消息。
20             【生产者发送消息时，消费者不需要处于运行状态】。
21             二、发布订阅模式：需要一个生产者发送消息到主题版块（Topic）中，可以有多个消费者订阅该版块来接受消息。
22             【生产者发送消息时，消费者必须处于运行状态，而且只能接受运行之后的消息】。
23     -->
24     <!-- 点对点模式 -->
25     <!-- <bean id="testQueue" class="org.apache.activemq.command.ActiveMQQueue">
26         <constructor-arg>
27             <value>test-queue</value>
28         </constructor-arg>
29     </bean> -->
30     <!-- 发布订阅模式 -->
31     <bean id="testTopic" class="org.apache.activemq.command.ActiveMQTopic">
32         <constructor-arg value="test-topic"/>
33     </bean>
34     
35     
36     <!-- 消费者 -->
37     <!-- 配置自定义消息监听器 -->
38     <bean id="myMessageListener" class="cn.soa.mq.MyMessageListener"></bean>
39     <!-- 配置MessageListenerContainer -->
40     <bean id="jmsContainer"
41         class="org.springframework.jms.listener.DefaultMessageListenerContainer">
42         <property name="connectionFactory" ref="connectionFactory"/>
43         <property name="destination" ref="testTopic"/>    <!-- 这儿注意生产者使用的是那种模式并且用哪个队列来发送消息的 -->
44         <property name="messageListener" ref="myMessageListener"/>
45     </bean>

步骤三：执行更新操作时，通知秘书去同步索引库、缓存等

 1 　　@Autowired
 2     private ItemMapper itemMapper;
 3     //消息队列
 4     @Autowired
 5     private JmsTemplate jmsTemplate;
 6     @Resource(name="testTopic")
 7     private Destination testTopic;

 8 　　@Override
 9     public int saveItem(Item item) {
10         final String itemId = UUIDUtils.getUUID();
11         if(StringUtil.isNullOrBlank(item.getId())){
12             //如果商品主键为空，则设置一个ID
13             item.setId(itemId);
14         }
15         if(StringUtil.isNullOrBlank(String.valueOf(item.getCreateTime()))){
16             //如果创建时间为空，则设置当前时间为创建时间
17             item.setCreateTime(DateTimeUtils.getCurrentDate());
18         }
19         int save = itemMapper.saveItem(item);
20         if(save == 1){
21             //如果新增商品成功，则发送商品ID到消息队列中，目的同步索引库、缓存等
22             jmsTemplate.send(testTopic, new MessageCreator(){
23                 @Override
24                 public Message createMessage(Session session) throws JMSException {
25                     // 将商品ID发送出去
26                     logger.error("发送新增商品的ID到MQ消息队列中：{}==============================");
27                     TextMessage message = session.createTextMessage(itemId);
28                     return message;
29                 }
30             });
31         }
32         return save;
33     }

步骤四：使用MQ监听器同步索引库(监听器需在spring配置文件中配置)

 1 public class MyMessageListener implements MessageListener{
 2     
 3     private final static Logger logger = LoggerFactory.getLogger(MyMessageListener.class);
 4     
 5     @Autowired
 6     private SolrServer solrServer;
 7     
 8     @Autowired
 9     private ItemService itemService;
10 
11     /**
12      * 根据监听到的商品ID来同步索引库数据。
13      */
14     @Override
15     public void onMessage(Message message) {
16         logger.info("============开始同步索引库================");
17         // 根据不同业务逻辑进行相应处理
18         if(message instanceof TextMessage){
19             try {
20                 TextMessage textMessage = (TextMessage) message;
21                 String ID = textMessage.getText();    //监听到新商品ID
22                 Item newItem = itemService.findItemById(ID);     //根据新主键查询到商品信息
23                 // 将商品数据封装到SolrInputDocument对象
24                 SolrInputDocument doc = new SolrInputDocument();
25                 doc.addField("id", newItem.getId());
26                 doc.addField("product_catalog_name", "忠哥系列");
27                 doc.addField("product_price", newItem.getPrice());
28                 doc.addField("product_name", newItem.getItemName());
29                 
30                 // 添加到索引库
31                 solrServer.add(doc);
32                 // 提交
33                 solrServer.commit();
34             } catch (Exception e) {
35                 logger.error("同步索引库失败：{}"+e.getMessage());
36             }
37         }
38     }
39 }