手册   游戏   WEB   字典   单词   在线工具  

代码:阿里云MQ JAVA-SDK简单封装

只是对阿里MQ的ajva-SDK做了个简单封装。

发布者直接调用：

封装类：MQProducerUtils

/**

/**

/**

* 指定时间发送MQ

* @param topic 可理解为一级类别

* @param tag 可理解为Gmail中的标签，对消息进行再归类，方便Consumer指定过滤条件在ONS服务器过滤

* @param body 消息体

* @param date 指定时间发送消息

* @return 消息的唯一ID 如果返回为null就表示发送失败，主流程不需要管返回

*/

监听者启动监听

MQConsumerListener

/**

* 集群订阅消息

* @param topic MQ主题

* @param tag  二级类别

* @param callBack 回调包名+类名

* @param consumerId 订阅者ID

*/

public static void MQListener(String topic, String tag, final MQCallBack callBack,String consumerId)

/**

* 广播的方式发送MQ

* @param topic MQ主题

* @param tag 二级类别

* @param callback 回调包名+类名

* @param consumerId 订阅者ID

*/

错误示例二：不同JVM同一订阅组（CID-ONS-FAQ）订阅的TOPIC相同，但Tag不同（分别NM-ONS-FAQ-1、NM-ONS-FAQ-2）

建议解决方案

请确保在不同JVM中使用相同的ConsumerId启动多个Consumer时，配置的Topic和Tag是一致的。

参见：

启动多个程序，CID topic tag 必须是一套

比方说您有四个程序启动

集群模式：（4个程序一起瓜分消息）

程序1：CID1 topic1 tag1

程序2：CID1 topic1 tag1

程序3：CID1 topic1 tag1

程序4：CID1 topic1 tag1

广播模式：（前面两个程序一组瓜分消息，后面两个程序一组瓜分消息，这两个组都能收到相同消息）

程序1：CID1 topic1 tag1

程序2：CID1 topic1 tag1

程序3：CID2 topic1 tag1

程序4：CID2 topic1 tag1

1. 代码

  1. import com.zcmall.core.config.AbstractFileConfig;
  2.  
  3. /**
  4.  * 邮件配置文件
  5.  * 
  6.  * @author wupengfei wupf86@126.com
  7.  * 
  8.  */
  9. public abstract class MQFileConfig extends AbstractFileConfig {
  10.  
  11.     private static final String FILE = "classpath*:config/mq/mq.properties";
  12.  
  13.     static {
  14.         load(FILE, getKeyPrefix());
  15.     }
  16.  
  17.     /**
  18.      * 获取指定键的值
  19.      * 
  20.      * @param config
  21.      * @return
  22.      */
  23.     public static String getValue(MQConfigEnum config) {
  24.         return AbstractFileConfig.getValue(getKeyPrefix() + config.getKey(), null);
  25.     }
  26.  
  27.     /**
  28.      * 获取配置文件中指定key的值，若==null则返回默认值
  29.      * 
  30.      * @param config
  31.      * @param defaultValue
  32.      * @return
  33.      */
  34.     public static String getValue(MQConfigEnum config, String defaultValue) {
  35.         return AbstractFileConfig.getValue(getKeyPrefix() + config.getKey(), defaultValue);
  36.     }
  37.  
  38.     /**
  39.      * 获取前辍
  40.      * 
  41.      * @return
  42.      */
  43.     public static String getKeyPrefix() {
  44.         return "mq#";
  45.     }
  46.  
  47.     /**
  48.      * 获取配置文件的路径
  49.      * 
  50.      * @return
  51.      */
  52.     public static String getFile() {
  53.         return FILE;
  54.     }
  55. }
  56.  
  57. /**
  58.  * 阿里MQ
  59.  *
  60.  * @author gjl gjl@163.com
  61.  *
  62.  */
  63. public enum MQConfigEnum {
  64.     /**
  65.      * 发送端ID
  66.      */
  67.     PRODUCER_ID("producer.id"),
  68.     /**
  69.      * 消费端ID
  70.      */
  71.     CONSUMER_ID("consumer.id"),
  72.     /**
  73.      * 阿里MQ配置
  74.      */
  75.     ACCESS_KEY("access.key"),
  76.     /**
  77.      * MQ订阅者
  78.      */
  79.     LISTENER("listener"),
  80.  
  81.     /**
  82.      * 阿里MQ配置
  83.      */
  84.     SECRET_KEY("secret.key");
  85.  
  86.  
  87.     private String key;
  88.  
  89.     private MQConfigEnum(String key) {
  90.         this.key = key;
  91.     }
  92.  
  93.     public String getKey() {
  94.         return key;
  95.     }
  96.  
  97. }
  98.  
  99. import com.aliyun.openservices.ons.api.Message;
  100. import com.aliyun.openservices.ons.api.ConsumeContext;
  101.  
  102. public interface MQCallBack {
  103.     public boolean exec(Message message, ConsumeContext context);
  104. }
  105.  
  106. import java.util.Properties;
  107.  
  108. import com.aliyun.openservices.ons.api.Action;
  109. import com.aliyun.openservices.ons.api.ConsumeContext;
  110. import com.aliyun.openservices.ons.api.Consumer;
  111. import com.aliyun.openservices.ons.api.Message;
  112. import com.aliyun.openservices.ons.api.MessageListener;
  113. import com.aliyun.openservices.ons.api.ONSFactory;
  114. import com.aliyun.openservices.ons.api.PropertyKeyConst;
  115. import com.aliyun.openservices.ons.api.PropertyValueConst;
  116. import com.zcmall.core.log.Log;
  117. import com.zcmall.core.log.LogFactory;
  118.  
  119. public class MQConsumerListener {
  120.     private static Log log = LogFactory.getLog(MQProducerUtils.class);
  121.  
  122.     private static Properties props;
  123.      
  124.     private static String LOG_INFO = "启动集群订阅：topic=%s,tag=%s,consumerId=%s,callBack=%s";
  125.      
  126.     static {
  127.         props = new Properties();
  128.         props.put(PropertyKeyConst.AccessKey, MQFileConfig.getValue(MQConfigEnum.ACCESS_KEY));
  129.         props.put(PropertyKeyConst.SecretKey, MQFileConfig.getValue(MQConfigEnum.SECRET_KEY));
  130.     }
  131.  
  132.     /**
  133.      * 集群订阅消息
  134.      * @param topic MQ主题
  135.      * @param tag  二级类别
  136.      * @param callBack 回调包名+类名
  137.      * @param consumerId 订阅者ID
  138.      */
  139.     public static void MQListener(String topic, String tag, final MQCallBack callBack,String consumerId){
  140.         props.put(PropertyKeyConst.ConsumerId, consumerId);
  141. //      props.put(PropertyKeyConst.MessageModel,PropertyValueConst.CLUSTERING);
  142.         Consumer consumer = ONSFactory.createConsumer(props);
  143.         System.out.println(consumer);
  144.         consumer.subscribe(topic, tag, new MessageListener() {
  145.             public Action consume(Message message, ConsumeContext context) {
  146.                 boolean bool = callBack.exec(message, context);
  147.                 Action result = Action.CommitMessage;
  148.                 if(!bool){
  149.                     result = Action.ReconsumeLater;
  150.                 }
  151.                 return result;
  152.             }
  153.         });
  154.         consumer.start();
  155.         log.info(String.format(LOG_INFO, topic,tag,consumerId,callBack));
  156.     }
  157.      
  158.     /**
  159.      * 广播的方式发送MQ
  160.      * @param topic MQ主题
  161.      * @param tag 二级类别
  162.      * @param callBack 回调包名+类名
  163.      * @param consumerId 订阅者ID
  164.      */
  165.     public static void MQPubSubListener(final String topic,final String tag, final MQCallBack callBack,String consumerId){
  166.         props.put(PropertyKeyConst.MessageModel,PropertyValueConst.BROADCASTING);
  167.         Consumer consumer = ONSFactory.createConsumer(props);
  168.         consumer.subscribe(topic, tag, new MessageListener() {
  169.             public Action consume(Message message, ConsumeContext context) {
  170.                 boolean bool = callBack.exec(message, context);
  171.                 Action result = Action.CommitMessage;
  172.                 if(!bool){
  173.                     result = Action.ReconsumeLater;
  174.                 }
  175.                 return result;
  176.             }
  177.         });
  178.         consumer.start();
  179.         log.info(String.format(LOG_INFO, topic,tag,consumerId,callBack));
  180.     }
  181. }
  182.  
  183. import java.util.Date;
  184. import java.util.Properties;
  185.  
  186. import com.aliyun.openservices.ons.api.Message;
  187. import com.aliyun.openservices.ons.api.ONSFactory;
  188. import com.aliyun.openservices.ons.api.Producer;
  189. import com.aliyun.openservices.ons.api.PropertyKeyConst;
  190. import com.aliyun.openservices.ons.api.SendResult;
  191. import com.zcmall.core.log.Log;
  192. import com.zcmall.core.log.LogFactory;
  193.  
  194. public class MQProducerUtils {
  195.  
  196.     private static Log log = LogFactory.getLog(MQProducerUtils.class);
  197.  
  198.     private static Properties props;
  199.  
  200.     static {
  201.         props = new Properties();
  202.         props.put(PropertyKeyConst.ProducerId, MQFileConfig.getValue(MQConfigEnum.PRODUCER_ID));
  203.         props.put(PropertyKeyConst.AccessKey, MQFileConfig.getValue(MQConfigEnum.ACCESS_KEY));
  204.         props.put(PropertyKeyConst.SecretKey, MQFileConfig.getValue(MQConfigEnum.SECRET_KEY));
  205.     }
  206.  
  207.     /**
  208.      * 发送集群MQ 多个消费者只会有一个消费者接收到。
  209.      * @param topic 可理解为一级类别
  210.      * @param tag 可理解为Gmail中的标签，对消息进行再归类，方便Consumer指定过滤条件在ONS服务器过滤
  211.      * @param body 消息体
  212.      * @return 消息的唯一ID 如果返回为null就表示发送失败，主流程不需要管返回
  213.      */
  214.     public static String sendMQ(String topic, String tag, String body) {
  215.         Message msg = new Message(topic, tag, body.getBytes());
  216.         return sendMQ(msg);
  217.     }
  218.      
  219.     /**
  220.      * 设置延迟发送MQ
  221.      * @param topic 可理解为一级类别
  222.      * @param tag 可理解为Gmail中的标签，对消息进行再归类，方便Consumer指定过滤条件在ONS服务器过滤
  223.      * @param body 消息体
  224.      * @param delayTime 延迟的时间单位毫秒
  225.      * @return 消息的唯一ID 如果返回为null就表示发送失败，主流程不需要管返回
  226.      */
  227.     public static String sendMQDelayTime(String topic, String tag, String body,long delayTime){
  228.         Message msg = new Message(topic, tag, body.getBytes());
  229.         msg.setStartDeliverTime(System.currentTimeMillis() + delayTime);
  230.         return sendMQ(msg);
  231.     }
  232.  
  233.     /**
  234.      * 指定时间发送MQ
  235.      * @param topic 可理解为一级类别
  236.      * @param tag 可理解为Gmail中的标签，对消息进行再归类，方便Consumer指定过滤条件在ONS服务器过滤
  237.      * @param body 消息体
  238.      * @param date 指定时间发送消息
  239.      * @return 消息的唯一ID 如果返回为null就表示发送失败，主流程不需要管返回
  240.      */
  241.     public static String sendMqDelayDate(String topic, String tag, String body,Date date){
  242.         Message msg = new Message(topic, tag, body.getBytes());
  243.         msg.setStartDeliverTime(date.getTime());
  244.         return sendMQ(msg);
  245.     }
  246.  
  247.     /**
  248.      * 发送消息
  249.      * @param msg
  250.      * @return
  251.      */
  252.     private static String sendMQ(Message msg) {
  253.         try {
  254.             Producer producer = ONSFactory.createProducer(props);
  255.             // 在发送消息前，必须调用start方法来启动Producer，只需调用一次即可。
  256.             producer.start();
  257.             // 发送消息，只要不抛异常就是成功
  258.             SendResult sendResult = producer.send(msg);
  259.             producer.shutdown();
  260.             return sendResult.getMessageId();
  261.         } catch (Exception e) {
  262.             // 吃掉这个异常不影响主流程
  263.             log.error("###########消息队列发送失败：" + e.getMessage());
  264.         }
  265.         return null;
  266.     }
  267. }

来源: https://www.oschina.net/code/snippet_2349285_58611


与本文相关文章

  1. Java 第一章 开发简单java应用程序 笔记
  2. java简单坦克大战制作代码
  3. Java 总结 - 封装继承多态
  4. JAVA JNI原理详细介绍及简单实例代码
  5. Java SDK 夯住 (Hang) 问题排查
  6. Java 9 揭秘（6. 封装模块）
  7. 详解 JAVA 的封装
  8. [直播预告] 阿里特邀专家关键: Java 无锁集合代码分析
暂无,快来抢沙发吧！
更多





黑ICP备17005318号-1