查看: 1287|回复: 0

[Java语言] 浅谈使用java实现阿里云消息队列简单封装

发表于 2018-4-16 08:00:05

一、前言

最近公司有使用阿里云消息队列的需求,为了更加方便使用,本人用了几天时间将消息队列封装成api调用方式以方便内部系统的调用,现在已经完成,特此记录其中过程和使用到的相关技术,与君共勉。

现在阿里云提供了两种消息服务:mns服务和ons服务,其中我认为mns是简化版的ons,而且mns的消息消费需要自定义轮询策略的,相比之下,ons的发布与订阅模式功能更加强大(比如相对于mns,ons提供了消息追踪、日志、监控等功能),其api使用起来更加方便,而且听闻阿里内部以后不再对mns进行新的开发,只做维护,ons服务则会逐步替代mns服务成为阿里消息服务的主打产品,所以,如果有使用消息队列的需求,建议不要再使用mns,使用ons是最好的选择。

涉及到的技术:Spring,反射、动态代理、Jackson序列化和反序列化

在看下面的文章之前,需要先看上面的文档以了解相关概念(Topic、Consumer、Producer、Tag等)以及文档中提供的简单的发送和接收代码实现。

该博文只针对有消息队列知识基础的朋友看,能帮上大家的忙我自然很高兴,看不懂的也不要骂,说明你路子不对。

二、设计方案

1.消息发送

在一个简单的cs架构中,假设server会监听一个Topic的Producer发送的消息,那么它首先应该提供client一个api,client只需要简单的调用该api,就可以通过producer来生产消息

2.消息接收

由于api是server制定的,所以server当然也知道如何消费这些消息

在这个过程中,server实际充当着消费者的角色,client实际充当着生产者的角色,但是生产者生产消息的规则则由消费者制定以满足消费者消费需求。

3.最终目标

我们要创建一个单独的jar包,起名为queue-core为生产者和消费者提供依赖和发布订阅的具体实现。

三、消息发送

1.消费者提供接口

  1. @Topic(name="kdyzm",producerId="kdyzm_producer")
  2. public interface UserQueueResource {
  3. @Tag("test1")
  4. public void handleUserInfo(@Body @Key("userInfoHandler") UserModel user);
  5. @Tag("test2")
  6. public void handleUserInfo1(@Body @Key("userInfoHandler1") UserModel user);
  7. }
复制代码

由于Topic和producer之间是N:1的关系,所以这里直接将producerId作为Topic的一个属性;Tag是一个很关键的过滤条件,消费者通过它进行消息的分类做不同的业务处理,所以,这里使用Tag作为路由条件。

2.生产者使用消费者提供的api发送消息

由于消费者只提供了接口给生产者使用,接口是没有办法直接使用的,因为没有办法实例化,这里使用动态代理生成对象,在消费者提供的api中,添加如下config,以方便生产者直接导入config即可使用,这里使用了基于java的spring config,请知悉。

  1. @Configuration
  2. public class QueueConfig {
  3. @Autowired
  4. @Bean
  5. public UserQueueResource userQueueResource() {
  6. return QueueResourceFactory.createProxyQueueResource(UserQueueResource.class);
  7. }
  8. }
复制代码

3.queue-core对生产者发送消息的封装

以上1中所有的注解(Topic、Tag、Body 、Key)以及2中使用到的QueueResourceFactory类都要在queue-core中定义,其中注解的定义只是定义了规则,真正的实现实际上是在QueueResourceFactory中

  1. import java.lang.reflect.InvocationHandler;
  2. import java.lang.reflect.Method;
  3. import java.lang.reflect.Proxy;
  4. import org.slf4j.Logger;
  5. import org.slf4j.LoggerFactory;
  6. import com.aliyun.openservices.ons.api.Message;
  7. import com.aliyun.openservices.ons.api.Producer;
  8. import com.aliyun.openservices.ons.api.SendResult;
  9. import com.wy.queue.core.api.MQConnection;
  10. import com.wy.queue.core.utils.JacksonSerializer;
  11. import com.wy.queue.core.utils.MQUtils;
  12. import com.wy.queue.core.utils.QueueCoreSpringUtils;
  13. public class QueueResourceFactory implements InvocationHandler {
  14. private static final Logger logger=LoggerFactory.getLogger(QueueResourceFactory.class);
  15. private String topicName;
  16. private String producerId;
  17. private JacksonSerializer serializer=new JacksonSerializer();
  18. private static final String PREFIX="PID_";
  19. public QueueResourceFactory(String topicName,String producerId) {
  20. this.topicName = topicName;
  21. this.producerId=producerId;
  22. }
  23. public static <T> T createProxyQueueResource(Class<T> clazz) {
  24. String topicName = MQUtils.getTopicName(clazz);
  25. String producerId = MQUtils.getProducerId(clazz);
  26. T target = (T) Proxy.newProxyInstance(QueueResourceFactory.class.getClassLoader(),
  27. new Class<?>[] { clazz }, new QueueResourceFactory(topicName,producerId));
  28. return target;
  29. }
  30. @Override
  31. public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
  32. if(args.length == 0 || args.length>1){
  33. throw new RuntimeException("only accept one param at queueResource interface.");
  34. }
  35. String tagName=MQUtils.getTagName(method);
  36. ProducerFactory producerFactory = QueueCoreSpringUtils.getBean(ProducerFactory.class);
  37. MQConnection connectionInfo = QueueCoreSpringUtils.getBean(MQConnection.class);
  38. Producer producer = producerFactory.createProducer(PREFIX+connectionInfo.getPrefix()+"_"+producerId);
  39. //发送消息
  40. Message msg = new Message( //
  41. // 在控制台创建的 Topic,即该消息所属的 Topic 名称
  42. connectionInfo.getPrefix()+"_"+topicName,
  43. // Message Tag,
  44. // 可理解为 Gmail 中的标签,对消息进行再归类,方便 Consumer 指定过滤条件在 MQ 服务器过滤
  45. tagName,
  46. // Message Body
  47. // 任何二进制形式的数据, MQ 不做任何干预,
  48. // 需要 Producer 与 Consumer 协商好一致的序列化和反序列化方式
  49. serializer.serialize(args[0]).getBytes());
  50. SendResult sendResult = producer.send(msg);
  51. logger.info("Send Message success. Message ID is: " + sendResult.getMessageId());
  52. return null;
  53. }
  54. }
复制代码

这里特意将自定义包和第三方使用的包名都贴过来了,以便于区分。

这里到底做了哪些事情呢?

发送消息的过程就是动态代理创建一个代理对象,该对象调用方法的时候会被拦截,首先解析所有的注解,比如topicName、producerId、tag等关键信息从注解中取出来,然后调用阿里sdk发送消息,过程很简单,但是注意,这里发送消息的时候是分环境的,一般来讲现在企业中会区分QA、staging、product三种环境,其中QA和staging是测试环境,对于消息队列来讲,也是会有三种环境的,但是QA和staging环境往往为了降低成本使用同一个阿里账号,所以创建的topic和productId会放到同一个区域下,这样同名的TopicName是不允许存在的,所以加上了环境前缀加以区分,比如QA_TopicName,PID_Staging_ProducerId等等;另外,queue-core提供了MQConnection接口,以获取配置信息,生产者服务只需要实现该接口即可。

4.生产者发送消息

  1. @Autowired
  2. private UserQueueResource userQueueResource;
  3. @Override
  4. public void sendMessage() {
  5. UserModel userModel=new UserModel();
  6. userModel.setName("kdyzm");
  7. userModel.setAge(25);
  8. userQueueResource.handleUserInfo(userModel);
  9. }
复制代码

只需要数行代码即可将消息发送到指定的Topic,相对于原生的发送代码,精简了太多。

四、消息消费

相对于消息发送,消息的消费要复杂一些。

1.消息消费设计

由于Topic和Consumer之间是N:N的关系,所以将ConsumerId放到消费者具体实现的方法上

  1. @Controller
  2. @QueueResource
  3. public class UserQueueResourceImpl implements UserQueueResource {
  4. private Logger logger = LoggerFactory.getLogger(this.getClass());
  5. @Override
  6. @ConsumerAnnotation("kdyzm_consumer")
  7. public void handleUserInfo(UserModel user) {
  8. logger.info("收到消息1:{}", new Gson().toJson(user));
  9. }
  10. @Override
  11. @ConsumerAnnotation("kdyzm_consumer1")
  12. public void handleUserInfo1(UserModel user) {
  13. logger.info("收到消息2:{}", new Gson().toJson(user));
  14. }
  15. }
复制代码

这里又有两个新的注解@QueueResource和@ConsumerAnnotation,这两个注解后续会讨论如何使用。有人会问我为什么要使用ConsumerAnnotation这个名字而不使用Consumer这个名字,因为Consumer这个名字和aliyun提供的sdk中的名字冲突了。。。。

在这里, 消费者提供api 接口给生产者以方便生产者发送消息,消费者则实现该接口以消费生产者发送的消息,如何实现api接口就实现了监听,这点是比较关键的逻辑。

2.queue-core实现消息队列监听核心逻辑

第一步:使用sping 容器的监听方法获取所有加上QueueResource注解的Bean

第二步:分发处理Bean

如何处理这些Bean呢,每个Bean实际上都是一个对象,有了对象,比如上面例子中的UserQueueResourceImpl 对象,我们可以拿到该对象实现的接口字节码对象,进而可以拿到该接口UserQueueRerousce上的注解以及方法上和方法中的注解,当然UserQueueResourceImpl实现方法上的注解也能拿得到,这里我将获取到的信息以consumerId为key,其余相关信息封装为Value缓存到了一个Map对象中,核心代码如下:

  1. Class<?> clazz = resourceImpl.getClass();
  2. Class<?> clazzIf = clazz.getInterfaces()[0];
  3. Method[] methods = clazz.getMethods();
  4. String topicName = MQUtils.getTopicName(clazzIf);
  5. for (Method m : methods) {
  6. ConsumerAnnotation consumerAnno = m.getAnnotation(ConsumerAnnotation.class);
  7. if (null == consumerAnno) {
  8. // logger.error("method={} need Consumer annotation.", m.getName());
  9. continue;
  10. }
  11. String consuerId = consumerAnno.value();
  12. if (StringUtils.isEmpty(consuerId)) {
  13. logger.error("method={} ConsumerId can't be null", m.getName());
  14. continue;
  15. }
  16. Class<?>[] parameterTypes = m.getParameterTypes();
  17. Method resourceIfMethod = null;
  18. try {
  19. resourceIfMethod = clazzIf.getMethod(m.getName(), parameterTypes);
  20. } catch (NoSuchMethodException | SecurityException e) {
  21. logger.error("can't find method={} at super interface={} .", m.getName(), clazzIf.getCanonicalName(),
  22. e);
  23. continue;
  24. }
  25. String tagName = MQUtils.getTagName(resourceIfMethod);
  26. consumersMap.put(consuerId, new MethodInfo(topicName, tagName, m));
  27. }
复制代码

第三步:通过反射实现消费的动作

首先,先确定好反射动作执行的时机,那就是监听到了新的消息

其次,如何执行反射动作?不赘述,有反射相关基础的童鞋都知道怎么做,核心代码如下所示:

  1. MQConnection connectionInfo = QueueCoreSpringUtils.getBean(MQConnection.class);
  2. String topicPrefix=connectionInfo.getPrefix()+"_";
  3. String consumerIdPrefix=PREFIX+connectionInfo.getPrefix()+"_";
  4. for(String consumerId:consumersMap.keySet()){
  5. MethodInfo methodInfo=consumersMap.get(consumerId);
  6. Properties connectionProperties=convertToProperties(connectionInfo);
  7. // 您在控制台创建的 Consumer ID
  8. connectionProperties.put(PropertyKeyConst.ConsumerId, consumerIdPrefix+consumerId);
  9. Consumer consumer = ONSFactory.createConsumer(connectionProperties);
  10. consumer.subscribe(topicPrefix+methodInfo.getTopicName(), methodInfo.getTagName(), new MessageListener() { //订阅多个Tag
  11. public Action consume(Message message, ConsumeContext context) {
  12. try {
  13. String messageBody=new String(message.getBody(),"UTF-8");
  14. logger.info("receive message from topic={},tag={},consumerId={},message={}",topicPrefix+methodInfo.getTopicName(),methodInfo.getTagName(),consumerIdPrefix+consumerId,messageBody);
  15. Method method=methodInfo.getMethod();
  16. Class<?> parameType = method.getParameterTypes()[0];
  17. Object arg = jacksonSerializer.deserialize(messageBody, parameType);
  18. Object[] args={arg};
  19. method.invoke(resourceImpl, args);
  20. } catch (Exception e) {
  21. logger.error("",e);
  22. }
  23. return Action.CommitMessage;
  24. }
  25. });
  26. consumer.start();
  27. logger.info("consumer={} has started.",consumerIdPrefix+consumerId);
  28. }
复制代码

五、完整代码见下面的git链接

https://github.com/kdyzm/queue-core.git

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持程序员之家。



回复

使用道具 举报