新功能发布记录
公告
DefaultMQProducer producer = new DefaultMQProducer(namespace, groupName,// ACL权限new AclClientRPCHook(new SessionCredentials(AK, SK)), true, null);
// 实例化消费者DefaultMQPushConsumer pushConsumer = new DefaultMQPushConsumer(NAMESPACE,groupName,new AclClientRPCHook(new SessionCredentials(AK, SK)),new AllocateMessageQueueAveragely(), true, null);
DefaultLitePullConsumer pullConsumer = new DefaultLitePullConsumer(NAMESPACE,groupName,new AclClientRPCHook(new SessionCredentials(AK, SK)));// 设置NameServer的地址pullConsumer.setNamesrvAddr(NAMESERVER);pullConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);pullConsumer.setAutoCommit(false);pullConsumer.setEnableMsgTrace(true);pullConsumer.setCustomizedTraceTopic(null);
package com.lazycece.sbac.rocketmq.messagemodel;import lombok.extern.slf4j.Slf4j;import org.apache.rocketmq.spring.annotation.MessageModel;import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;import org.apache.rocketmq.spring.core.RocketMQListener;import org.springframework.stereotype.Component;/*** @author lazycece* @date 2019/8/21*/@Slf4j@Componentpublic class MessageModelConsumer {@Component@RocketMQMessageListener(topic = "topic-message-model",consumerGroup = "message-model-consumer-group",enableMsgTrace = true,messageModel = MessageModel.CLUSTERING)public class ConsumerOne implements RocketMQListener<String> {@Overridepublic void onMessage(String message) {log.info("ConsumerOne: {}", message);}}}





文档反馈