整体流程:

消息变化-->zookeeper监听-->metaq集群服务器--->通过filter过滤--->消息接收者-->执行操作.

一些配置:

发送者:

引入datachange-client

<dependency><groupId>com.pan.datachange</groupId><artifactId>pan-datachange-client</artifactId><version>1.0.0-SNAPSHOT</version>
</dependency>

环境配置:

  <!-- 加载DataChange的环境变量 --><beanclass="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer"><!-- webserver环境 --><property name="location"value="file:c:/dataChange-receiver1.properties" /><property name="ignoreResourceNotFound" value="true" /><property name="ignoreUnresolvablePlaceholders" value="true" /><!-- dataChange配置 --><property name="properties" ref="localDataChangeProperties" /></bean><bean name="localDataChangeProperties" class="org.springframework.beans.factory.config.PropertiesFactoryBean"><property name="location" value="classpath:exampleConfigDefault/dataChange-receiver.properties" /></bean>

dataChange-receiver.properties

dataChange.zkConnect=127.0.0.1:2181
dataChange.zkSessionTimeoutMs=3000
dataChange.zkConnectionTimeoutMs=10000
dataChange.zkSyncTimeMs=3000
dataChange.sender=testSenderdataChangeThreadExecutor.poolSize=10

然后发送方法:

@Resource
@Qualifier("defaultDataChangeManager")
private DataChangeManager dataChangeManager ;public void send(){DataChangeDO dataChangeDO = new DataChangeDO();dataChangeDO.setDataId(200);//过滤规则dataChangeDO.setDataType(DataType.member);dataChangeDO.setChangeType(ChangeTypeConstant.DATA_INSERT);dataChangeManager.sendMessage(dataChangeDO);}

接收者配置:

一样的需要引入dataChange-client

<dependency><groupId>com.pan.datachange</groupId><artifactId>pan-datachange-client</artifactId><version>1.0.0-SNAPSHOT</version>
</dependency>

实现dataChange的DataChangeListener接口

public class MessageConsumer implements DataChangeListener {public void receiveMessage(DataChangeDO dataChangeDO) {System.out.println("consumerMessage:" + dataChangeDO);}
}

过滤配置:

1.dataChangeHandleMapping配置:

首先实现dataChangeListener接口

public class ReceivedDataChangeListener implements DataChangeListener {protected Logger logger = LoggerFactory.getLogger(getClass());protected ExecutorService dataChangePool = Executors.newFixedThreadPool(100);public void receiveMessage(DataChangeDO dataChangeDO) {//接收消息进行的一些操作}@PreDestroypublic void destroy() {dataChangePool.shutdown();}
}

然后在配置bean:

<bean name="receivedDataChangeHandleMapping" class="com.pan.datachange.client.domain.ChangeHandleMapping"><property name="sender" value="发送者"></property><property name="receiver" value="接收者"></property><property name="dataChangeListener"><!--这里加载的是自己实现DataChangeListener接口的类><bean class="com.pan.jcob.msgcenterjob.datachange.ReceivedDataChangeListener" /></property><!--过滤规则,只有符合这种规则的数据才接收--><property name="filterConfigList"><list><bean class="com.pan.datachange.common.domain.FilterConfig"><property name="dataType" value="过滤格式"></property><property name="changeType" value="变化类型"></property></bean></list></property></bean>

2.ReceivedManager配置:

 <bean name="receiveManager" class="com.pan.datachange.client.receive.ReceiveManager" destroy-method="destroy"><property name="zkDataChangeRoot" value="/datachange/metaq" /><property name="zkConnect" value="zkAddr" /><property name="changeHandleMappingList"><list><!-- 一个topic一个mapping，这里就是填写之前的bean --><ref bean="receivedDataChangeHandleMapping"></ref> </list></property><!-- 只处理发送时间在这个日期之后的消息，这样可以防止新应用接收到很久以前的消息 --><property name="acceptMsgUpdateTimeAfterThis"><bean factory-bean="dataChangeDateFormat" factory-method="parse"><constructor-arg value="2015-12-16" /></bean></property><!--配置不接收消息的bean--><property name="bConsumeFromMaxOffset" value="true" /></bean>