Просмотр исходного кода

解决阿里云rocketmq接入bug

release_toaliyun_real
xiaohanzi 6 лет назад
Родитель
Сommit
2f343f1495
2 измененных файлов: 5 добавлений и 59 удалений
  1. +3
    -29
      mallinkMQConsumer/src/main/java/com/iformall/mq/impl/AliyunRocketMqConsumer.java
  2. +2
    -30
      mallinkService/src/main/java/com/iformall/mq/impl/AliyunRocketMqMessageProducer.java

+ 3
- 29
mallinkMQConsumer/src/main/java/com/iformall/mq/impl/AliyunRocketMqConsumer.java Просмотреть файл

@@ -65,35 +65,9 @@ public class AliyunRocketMqConsumer extends MqBaseConsumer {
return Action.CommitMessage;
}
});
consumer.start();
log.info(" aliyunrocketMq consumer start success!");
}
// @PostConstruct
// public void init() {
// Properties properties = new Properties();
// // 您在控制台创建的 Group ID
// properties.put(PropertyKeyConst.GROUP_ID, consumerGroupId);
// // 鉴权用 AccessKeyId,在阿里云服务器管理控制台创建
// properties.put(PropertyKeyConst.AccessKey, accessKeyId);
// // 鉴权用 AccessKeySecret,在阿里云服务器管理控制台创建
// properties.put(PropertyKeyConst.SecretKey, accessKeySecret);
// // 设置 TCP 接入域名,进入控制台的实例详情页面,在页面上方选择实例后,在实例信息中的“获取接入点信息”区域查看
// properties.put(PropertyKeyConst.NAMESRV_ADDR,namesrvAddr);
//
// Consumer consumer = ONSFactory.createConsumer(properties);
// consumer.subscribe("topic-1", "*", new MessageListener() {
// public Action consume(Message message, ConsumeContext context) {
// System.out.println("Receive: " + message);
// String msg = new String(message.getBody());
// doMessage(msg);
// return Action.CommitMessage;
// }
// });
// consumer.start();
// System.out.println("Consumer Started");
// }
// public void onMessage(String message) {
// doMessage(message);
// }
}

+ 2
- 30
mallinkService/src/main/java/com/iformall/mq/impl/AliyunRocketMqMessageProducer.java Просмотреть файл

@@ -27,7 +27,7 @@ import com.iformall.utils.JsonUtil;
* @Date: 2019-01-31
* @Description: RocketMQ消息生产者
*/
@Service
//@Service
@Profile(MQConfig.Impl.ALIYUN_ROCKET_MQ)
public class AliyunRocketMqMessageProducer implements MqBaseProducer {
private final Logger log = LoggerFactory.getLogger(this.getClass());
@@ -53,7 +53,7 @@ public class AliyunRocketMqMessageProducer implements MqBaseProducer {

private static Producer producer = null;
@PostConstruct
// @PostConstruct
public void init() {
Properties properties = new Properties();
properties.setProperty(PropertyKeyConst.GROUP_ID, groupId);
@@ -103,33 +103,5 @@ public class AliyunRocketMqMessageProducer implements MqBaseProducer {
// 在应用退出前,销毁 Producer 对象
// 注意:如果不销毁也没有问题
//producer.shutdown();
// try {
// data.setUuid(UUID.randomUUID().toString());
// wxMsgRecordService.save(data);
// Message msg = new Message( //
// // 在控制台创建的 Topic,即该消息所属的 Topic 名称
// topic,
// // Message Tag,
// // 可理解为 Gmail 中的标签,对消息进行再归类,方便 Consumer 指定过滤条件在消息队列 RocketMQ 版服务器过滤
// tags,
// // Message Body
// // 任何二进制形式的数据,消息队列 RocketMQ 版不做任何干预,
// // 需要 Producer 与 Consumer 协商好一致的序列化和反序列化方式
// JsonUtil.obj2Json(data).getBytes());
// // 设置代表消息的业务关键属性,请尽可能全局唯一,以方便您在无法正常收到消息情况下,可通过控制台查询消息并补发
// // 注意:不设置也不会影响消息正常收发
// msg.setKey(keys);
// // 发送消息,只要不抛异常就是成功
// // 打印 Message ID,以便用于消息发送状态查询
// SendResult sendResult = getProducer(producerGroupId).send(msg);
// //System.out.println("Send Message success. Message ID is: " + sendResult.getMessageId());
// data.setMsgStatus(EnumMsgRecordStatus.SEND_SUCC.getCode());
// wxMsgRecordMapper.updateByBaseMsg(data);
// } catch (Exception e) {
// log.error("Message Producer: Send Message Error ", e);
// }

}

}

Загрузка…
Отмена
Сохранить