Activemq实现Mysql与SolrCloud同步策略

向右看齐 2022-06-09 12:19 107阅读 0赞

应用场景：

商品数量很多，搜索功能为了保证用户体验，就使用了solrcloud，可是后面添加的商品可以入库，如何导入到solr库当中呢。我们用到了mq实现。再插入商品之后，发送一个商品id到队列上面去，是生产者。solr服务负责消费，监听商品添加消息，接收消息后，将对应的商品信息同步到索引库。

代码：

1.生产者 编码

@Autowired
    private JmsTemplate jmsTemplate;
    
    @Resource
    private Destination topicDestination;
    
    @Override
    	public E3Result addItem(TbItem tbItem, String desc) {
    		//生成商品ID
    		final long  itemid=IDUtils.genItemId();
    		//补全item的属性
    		tbItem.setId(itemid);
    //		商品状态，1-正常，2-下架，3-删除
    		tbItem.setStatus((byte)1);
    		tbItem.setCreated(new Date());
    		tbItem.setUpdated(new Date());
    		
    		//向商品表插入数据
    		itemMapper.insert(tbItem);
    		//创建一个商品描述表pojo
    		TbItemDesc itemDesc=new TbItemDesc();
    		//商品id
    		itemDesc.setItemId(itemid);
    		itemDesc.setItemDesc(desc);
    		itemDesc.setCreated(new Date());
    		itemDesc.setUpdated(new Date());
    		//向商品描述表插入数据
    		itemDescMapper.insert(itemDesc);
    		//发送id到activemq队列上面去
    		jmsTemplate.send(topicDestination,new MessageCreator(){
    			
    			@Override
    			public Message createMessage(Session session) throws JMSException {
    				TextMessage textMessage=session.createTextMessage(itemid+"");
    				return textMessage;
    			}
    		});
    		//返回成功
    		return E3Result.ok();
    	}

2.生产者配置整合Spring，利用JMS工具发送接收消息

<?xml version="1.0" encoding="UTF-8"?>
    <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:dubbo="http://code.alibabatech.com/schema/dubbo" 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://code.alibabatech.com/schema/dubbo http://code.alibabatech.com/schema/dubbo/dubbo.xsd
    	http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-4.2.xsd">
    
    	
    	<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
    		<property name="brokerURL" value="tcp://192.168.139.128: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="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="itemAddTopic" />
    	</bean>
    </beans>

3.消费者编码

package cn.e3mall.search.message;
    
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.TextMessage;
    
    import org.apache.solr.client.solrj.SolrServer;
    import org.apache.solr.common.SolrInputDocument;
    import org.springframework.beans.factory.annotation.Autowired;
    
    import cn.e3mall.common.pojo.SearchItem;
    import cn.e3mall.search.mapper.ItemMapper;
    
    /**
     * 监听商品添加消息，接收消息后，将对应的商品信息同步到索引库
     */
    public class ItemAddMessageListener implements MessageListener {
    	
    	@Autowired
    	private ItemMapper itemMapper;
    	@Autowired
    	private SolrServer solrServer;
    
    	@Override
    	public void onMessage(Message message) {
    		try {
    			//从消息中取商品id
    			TextMessage textMessage = (TextMessage) message;
    			String text = textMessage.getText();
    			//从队列中获取了id
    			Long itemId = new Long(text);
    			//等待事务提交
    			Thread.sleep(1000);
    			//根据商品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());
    			//把文档写入索引库
    			solrServer.add(document);
    			//提交
    			solrServer.commit();
    		} catch (Exception e) {
    			e.printStackTrace();
    		}
    
    	}
    
    }

4.消费者配置

<?xml version="1.0" encoding="UTF-8"?>
    <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">
    	
    	
    	<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
    		<property name="brokerURL" value="tcp://192.168.139.128:61616" />
    	</bean>
    	
    	<bean id="connectionFactory"
    		class="org.springframework.jms.connection.SingleConnectionFactory">
    		
    		<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="itemAddTopic" />
    	</bean>
    	<bean id="myMessageListener" class="cn.e3mall.search.message.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="itemAddMessageListener" class="cn.e3mall.search.message.ItemAddMessageListener"/>
    	<bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
    		<property name="connectionFactory" ref="connectionFactory" />
    		<property name="destination" ref="topicDestination" />
    		<property name="messageListener" ref="itemAddMessageListener" />
    	</bean>
    </beans>

后来我们商品详情页面使用Freemarker静态页面，第一次生成的时候会比较慢，于是采用第一次添加商品的时候就去生成静态页面，所以，这个服务也需要订阅监听商品添加消息。这种订阅方式是主题订阅，也就是一对多的关系，一个生产者，一堆消费者。

<?xml version="1.0" encoding="UTF-8"?>
    <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">
    	
    	
    	<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
    		<property name="brokerURL" value="tcp://192.168.139.128:61616" />
    	</bean>
    	
    	<bean id="connectionFactory"
    		class="org.springframework.jms.connection.SingleConnectionFactory">
    		
    		<property name="targetConnectionFactory" ref="targetConnectionFactory" />
    	</bean>
    	
    	<bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">
    		<constructor-arg value="itemAddTopic" />
    	</bean>
    	
    	<bean id="htmlGenListener" class="cn.e3mall.item.listener.HtmlGenListener"/>
    	<bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
    		<property name="connectionFactory" ref="connectionFactory" />
    		<property name="destination" ref="topicDestination" />
    		<property name="messageListener" ref="htmlGenListener" />
    	</bean>
    </beans>

package cn.e3mall.item.controller;
    
    import java.io.FileWriter;
    import java.io.IOException;
    import java.io.Writer;
    import java.util.HashMap;
    import java.util.Map;
    
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.TextMessage;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.web.servlet.view.freemarker.FreeMarkerConfigurer;
    
    import cn.e3mall.item.pojo.Item;
    import cn.e3mall.pojo.TbItem;
    import cn.e3mall.pojo.TbItemDesc;
    import cn.e3mall.service.ItemService;
    import freemarker.template.Configuration;
    import freemarker.template.Template;
    
    /**
     * 监听商品添加消息
     * @author Julie
     *
     */
    public class HtmlGenListener implements MessageListener {
    
    	@Autowired
    	public ItemService itemService;
    	
    	@Autowired
    	private FreeMarkerConfigurer freeMarckerConfigurer;
    	
    	@Value("{HTML_GEN_PATH}")
    	private String HTML_GEN_PATH;
    	@Override
    	public void onMessage(Message message) {
    //		创建一个模板
    //		从消息中取得商品id
    		try {
    			TextMessage textMessage=(TextMessage)message;
    			String text=textMessage.getText();
    			Long itemId=new Long(text);
    			//等待事务提交
    			Thread.sleep(1000);
    //		根据商品id查询商品信息，商品基本信息和商品描述
    			TbItem tbItem=itemService.getItemById(itemId);
    			Item item = new Item(tbItem);
    			TbItemDesc itemDesc=itemService.getItemDesc(itemId);
    //		创建一个数据集，吧商品数据封装。
    			Map dataMap=new HashMap<>();
    			dataMap.put("item", item);
    			dataMap.put("itemDesc", itemDesc);
    			Configuration configuration=freeMarckerConfigurer.getConfiguration();
    //		加载模板对象
    			Template template= configuration.getTemplate("item.ftl");
    //		创建一个输出流，指定输出的目录以及文件名
    			Writer out=new FileWriter(HTML_GEN_PATH+itemId+".html");
    //		生成静态页面
    			template.process(dataMap, out);
    			
    //		关闭流
    			out.close();
    		} catch (Exception e) {
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		}
    	}
    
    }

总结：

mq的应用是系统应用的解耦，就算我添加solr库不成功，起码我的商品信息已经保存到数据库当中了。关于mq的应用场景还有：应用解耦，流量削锋和消息通讯等。