rocketMq实战(2)-客户端集成 博客分类: rocketmq rocketmq客户端
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文档和学习资料。
推荐阅读
-
rocketMq实战(4)-集群配置 博客分类: rocketmq rocketmq集群配置
-
rocketMq实战(4)-集群配置 博客分类: rocketmq rocketmq集群配置
-
rocketMq实战(2)-客户端集成 博客分类: rocketmq rocketmq客户端
-
rocketMq实战(5)-本地调试broker源码 博客分类: rocketmq rocketmqdebug
-
rocketMq实战(5)-本地调试broker源码 博客分类: rocketmq rocketmqdebug
-
rocketMq实战(3)-console和运维 博客分类: rocketmq rocketmq-console
-
rocketMq实战(2)-客户端集成