rocketMq实战(2)-客户端集成

rocketmq的配置和demo搞定后，离上线运用 还有很多问题要解决

如： 怎样集成 到项目中，并做到规范,易用。使用中有哪些问题是需要考虑的，监控运维问题怎么解决。

本文先解决 客户端集成，下面贴出我经过反复试验后的最终代码和配置。

 本文贴出的代码和配置都是经过反复测试和验证，并在实际项目中使用的，目前只使用几个重要的参数，更精细的配置请参考官方文档。

 pom.xml

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">  
  <modelVersion>4.0.0</modelVersion>
  <groupId>rockmqtest</groupId>
  <artifactId>rockmqtest</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  	<properties>
		<spring.version>3.1.1.RELEASE</spring.version>
		<slf4j.version>1.7.13</slf4j.version>
	</properties>
  <dependencies>
    <!-- RocketMQ Java SDK -->
<dependency>
    <groupId>com.alibaba.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>3.5.8</version>
</dependency>
	<dependency>
	<groupId>org.springframework</groupId>
	 <artifactId>spring-context</artifactId>
	 <version>3.1.1.RELEASE</version>
	</dependency>
	<dependency>
	<groupId>org.springframework</groupId>
	 <artifactId>spring-test</artifactId>
	 <version>3.1.1.RELEASE</version>
	</dependency>
	
<dependency>
    <groupId>ch.ethz.ganymed</groupId>
    <artifactId>ganymed-ssh2</artifactId>
    <version>build209</version>
</dependency>
<dependency>
    <groupId>commons-lang</groupId>
    <artifactId>commons-lang</artifactId>
    <version>2.6</version>
</dependency>
<dependency>
    <groupId>com.jcraft</groupId>
    <artifactId>jsch</artifactId>
    <version>0.1.46</version>
</dependency>

 
  	<dependency>
			<groupId>junit</groupId>
			<artifactId>junit</artifactId>
			<version>4.12</version>
			<scope>test</scope>
		</dependency>
 
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>jcl-over-slf4j</artifactId>
        <version>1.5.8</version>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-api</artifactId>
        <version>1.5.8</version>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-log4j12</artifactId>
        <version>1.5.8</version>
    </dependency>
    <dependency>
        <groupId>log4j</groupId>
        <artifactId>log4j</artifactId>
        <version>1.2.14</version>
    </dependency>
  </dependencies>
 
</project>

 

配置：

 

<!-- 同一个主题共用一个producer 不同主题不共用，防止发送速度快造成的性能问题 groupName= topic+producer-->
<bean id="mqProducer"   class="com.sfbest.rocketmq.online.MQProducer" init-method="start">
<constructor-arg>
<value>FirstGroupName</value>
</constructor-arg>
  <!-- 目标的主题 -->
  <property name="topic" value="TopicTest1"></property>
<property name="namesrvAddr" value="10.103.16.77:9876"></property>
</bean>
 

<!--  rocketMq 消费者配置 默认集群模式消费  -->
<!--  一个topic代表一个大业务 tag代表小业务如updateA,updateB。 同一个topic 共用一个consumer   consumerGroupName=topic+Consumer-->
<bean id="mqPushConsumer" class="com.sfbest.rocketmq.online.MQConsumer" init-method="init">
<!-- 消费者分组名，同一个业务分组名相同，不同则分组名不同 -->
<property name="consumerGroupName" value="firstSpringConsumer"></property>
  <!-- nameserver 地址和端口号 -->
  <property name="namesrvAddr" value="10.103.16.77:9876"></property>
  <!-- 订阅的话题 -->
  <property name="topic" value="TopicTest1"></property>
  <!-- 订阅话题的标记 *号表示所有TAG-->
  <property name="tagExpr"  value="*"></property>
 
	<property name="mqBusinessMap">
		 <map>
		    <entry key="TagA" value-ref="mqBusinessTestA"/>
		    <entry key="TagB" value-ref="mqBusinessTestB"/>
		 </map>
	</property>
</bean>

<!-- 自定义   具体业务实现类-->
<bean id="mqBusinessTestA" class="com.sfbest.rocketmq.online.MQBusinessTestAImpl"></bean>
<bean id="mqBusinessTestB" class="com.sfbest.rocketmq.online.MQBusinessTestBImpl"></bean>

 

 客户端 实现要考虑，扩展性，易用性，性能，规范。并且要根据rocketMq的特性编写,下面就讲讲我的实现。

关键参数可配置，设计命名规范和接口规范。

同一个主题共用一个producer 不同主题不共用，防止发送速度快造成的性能问题。

同一个主题共用一个cousumer 不用的tag实现分发，预留接口给业务实现。

 

 

Producer客户端公共类

 

/**
 * defaultMQProducer装饰器
 * 封装易用的发送方法
 * @author chenchangqun
 *
 */
public class BestMQProducer  extends DefaultMQProducer{
	private String topic;
	
	 private static final Logger LOG = LoggerFactory.getLogger(BestMQConsumer.class);

	public BestMQProducer(String producerGroupName){
		super(producerGroupName);
		//默认参数设置
		
	}
	
	
	public void init(){
		try {
			super.start();
			
		} catch (MQClientException e) {
			LOG.error("rockmq producer start fail",e);
		}
	}
	/**
	 * 发送方法
	 * @author chenchangqun
	 * @param key
	 * @param content
	 * @return
	 */
	public boolean sendMsg(String tag,String key,String content){

	      Message msg=null;  // body
		try {
			msg = new Message(topic, // topic
					tag, // tag
					key, // key
					content.getBytes("UTF-8"));
		} catch (Exception e) {
			LOG.error("create message fail ,cannot send msg ,invoke end 。topic={},tag={},key={},content={}",topic,tag,key,content,e);
			return false;
		}
	      try {
	    	  //没有初始化，则初始化
	    	  if(this.defaultMQProducerImpl.getServiceState()==ServiceState.CREATE_JUST){
	    		  LOG.info("mqProducer should be start");
	    		  this.init();
	    	  }
	    	 LOG.debug("mq send param tag:{},key:{},content:{}",tag,key,content);
			SendResult sendResult = super.send(msg);
			LOG.debug("sendResult:{}",sendResult.getSendStatus().toString());
//			sendResult.getSendStatus()==SendStatus.
//		     System.out.println("sendResult:{}"+sendResult);
			return sendResult.getSendStatus()==SendStatus.SEND_OK;
		} catch (MQClientException e) {
			LOG.error("send fail",e);
		} catch (RemotingException e) {
			LOG.error("send fail",e);
		} catch (MQBrokerException e) {
			LOG.error("send fail",e);
		} catch (InterruptedException e) {
			LOG.error("send fail",e);
		}	  
	      return false;
	}

	public String getTopic() {
		return topic;
	}

	public void setTopic(String topic) {
		this.topic = topic;
	}


 
	 
}

 producer的实现有两点考虑 

 

（1）覆盖 父类的 start,方便以后在启动时加业务逻辑。

（2）提供规范的发送方法，参数 包含 tag,key,content,因为同一个produder的tocpic是确定的。剩下的参数需要传递。可否提供topic参数？我认为不应该，如果以后发送方法多了，直接传递topic会导致代码可读性差，耦合性高。也可能会导致性能问题。

 

Consumer客户端公共类

 

 

/**
 * mq单例工厂类，设置了一些默认参数
 * 
 * @author chenchangqun
 *
 */
public class BaseConsumer {
	private String nameServAddr;
 
   private static DefaultMQPushConsumer consumer ;
   private BaseConsumer() {
   }
   public static DefaultMQPushConsumer getDefaultMQPushConsumer(String consumerName){ 
	   
	   if(consumer==null){
		   consumer = new DefaultMQPushConsumer(consumerName);          
		   consumer.setMessageModel(MessageModel.CLUSTERING);
		   consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
	   }

       return consumer;        
   }
   
   public String getNameServAddr() {
	return nameServAddr;
}
public void setNameServAddr(String nameServAddr) {
	this.nameServAddr = nameServAddr;
}
}

 

 

注意下面的部分

 

consumer.setMessageModel(MessageModel.CLUSTERING);

rocketMQ的重要特性就是支持集群消费，什么是集群消费，就是一个业务有多个实现，我们通常希望 其中一个实例消费成功后，其他实例就不要再消费了。而上述配置就是集群消费。

 

consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);

先解释个思维误区：如果一个消息被我的应用成功消费，是不是这条消息就没用了，可以干掉或丢弃。 答案：不对，在broker来讲你只是其中一个订阅者，这条消息可能被其他消费者订阅，broker只能记录你的消费进度。

offset是重要的名称，标记消息位置

我怎么知道消费了多少，还有多少消息没消费？答案：broker没统计这个，只能通过offset相减计算，即目前生产的   offset减去last cosumet offset。

 

要从上次消费的位置开始消费，配置如上。

 

Consumer客户端公共类

写道

/**
* mq参数设置和初始化
* 创建监听器，并根据不同的TAG分发
* @author chenchangqun
*
*/
public class BestMQConsumer {
private static final Logger LOG = LoggerFactory.getLogger(BestMQConsumer.class);
private String topic;
private String tagExpr;
private String namesrvAddr;
private Map<String,IMQBusiness> mqBusinessMap;
private String consumerGroupName;
//int count =0;
public void init(){
// 获取消息生产者
DefaultMQPushConsumer consumer = BaseConsumer.getDefaultMQPushConsumer(consumerGroupName);
// 订阅主体
try {
consumer.setNamesrvAddr(namesrvAddr);
consumer.subscribe(topic, tagExpr);
consumer.registerMessageListener( new MessageListenerConcurrently(){
//一次消费多条msg，如果第一条成功 第二条失败 返回 成功 失败？
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
for(MessageExt msg:msgs){
if(msg.getTags()!=null){
IMQBusiness mqBusiness= mqBusinessMap.get(msg.getTags());
if(mqBusiness!=null){
if(!mqBusiness.invokeMessage(msg)){
LOG.error("cosume fail return RECONSUME_LATER");
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
}


}
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

}
}
);

/**
* Consumer对象在使用之前必须要调用start初始化，初始化一次即可<br>
*/
consumer.start();
System.out.println("spring Consumer first Started.");
} catch (MQClientException e) {
LOG.error("cousumner init fail",e);
}
}
public void setTopic(String topic) {
this.topic = topic;
}
public void setTagExpr(String tagExpr) {
this.tagExpr = tagExpr;
}
public void setNamesrvAddr(String namesrvAddr) {
this.namesrvAddr = namesrvAddr;
}
public void setConsumerGroupName(String consumerGroupName) {
this.consumerGroupName = consumerGroupName;
}

public void setMqBusinessMap(Map<String, IMQBusiness> mqBusinessMap) {
this.mqBusinessMap = mqBusinessMap;
}


}

 

接口类

 

/**
 *消费MQ的业务接口
 * @author chenchangqun
 *
 */
public interface IMQBusiness {

	public boolean invokeMessage(MessageExt msg);
	
	
}

 

 

接口实现类

 

public class MQBusinessTestAImpl implements IMQBusiness {
	@Override
	public boolean invokeMessage(MessageExt msg) {
		System.out.println(" invoke A rec:"+new String(msg.getBody()));
		return true;
	}
}

 

 

cousumer使用了PUSH的方式，即被动通知。实际上实现是用到了长连接。

上面的cousumer代码 主要有两个重点，一个缺陷

（1）经过测试如果不返回CONSUME_SUCCESS，则会一直重试。

（2）实现了基于tag的分发，方便业务自定义实现。

缺陷

    没有配置每次消费条数，使用默认每次消费一条，如果一次消费多条msg，如果第一条成功 第二条失败 返回 成功 失败？，这种情况还需要根据业务具体情况处理

 

注意看我的附件，里面包含完整的代码和测试类。而电子书中的文档是我费劲心血收集的，应该是目前比较好的rocketmq文档和学习资料。