pom依赖
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>4.8.0</version>
</dependency>
一、同步发送消息
public static void main(String[] args) throws Exception {
// 创建一个producer，参数为Producer group ID
DefaultMQProducer producer = new DefaultMQProducer("money");
// 指定nameServer地址
producer.setNamesrvAddr("192.168.200.132:9876");
// 设置当发送失败时重试发送的次数，默认为2次
producer.setRetryTimesWhenSendFailed(3);
// 设置发送超时时效为5s，默认为3s
producer.setSendMsgTimeout(5000);
// 开启生产者
producer.start();
// 生产并发送100条消息
for (int i = 0; i < 100; i++) {
byte[] body = ("money" + i).getBytes();
Message message = new Message("moneyTopic", "moneyTag", body);
// 为消息指定key
message.setKeys("moneyKey" + i);
// 发送消息
SendResult sendResult = producer.send(message);
System.out.println(sendResult);
}
// 关闭producer
producer.shutdown();
}
二、异步发送消息
public static void main(String[] args) throws Exception {
// 创建一个producer，参数为Producer group ID
DefaultMQProducer producer = new DefaultMQProducer("money");
// 指定nameServer地址
producer.setNamesrvAddr("192.168.200.132:9876");
// 指定异步发送失败后不进行重试发送
producer.setRetryTimesWhenSendAsyncFailed(0);
// 指定新创建的Topic数量为2，默认为4
producer.setDefaultTopicQueueNums(2);
// 开启生产者
producer.start();
// 生产并发送100条消息
for (int i = 0; i < 100; i++) {
byte[] body = ("money" + i).getBytes();
Message message = new Message("moneyAsynTopicA", "moneyTag", body);
// 为消息指定key
message.setKeys("moneyKey" + i);
// 发送消息，指定回调
producer.send(message, new SendCallback() {
// 当producer接收到MQ发送来的ACK后就会触发该回调方法的执行
@Override
public void onSuccess(SendResult sendResult) {
System.out.println(sendResult);
}
@Override
public void onException(Throwable throwable) {
throwable.printStackTrace();
}
});
}
// 由于采取的异步，还没发送完就关了，所以修改休眠一会
TimeUnit.SECONDS.sleep(5);
// 关闭producer
producer.shutdown();
}
三、consumer消费消息
public static void main(String[] args) throws Exception {
// 定义一个pull消息
// DefaultLitePullConsumer consumer = new DefaultLitePullConsumer("consumerGroup");
// 定义一个push消费者
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumerGroup");
// 指定nameServer地址
consumer.setNamesrvAddr("192.168.200.132:9876");
// 指定从第一条消息开始消费
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
// 指定消费topic与tag
consumer.subscribe("moneyTopic", "*");
// 指定采用“广播模式”，进行消费，默认为“集群模式”
consumer.setMessageModel(MessageModel.BROADCASTING);
// 注册消息监听器
consumer.registerMessageListener(new MessageListenerConcurrently() {
// 一但broker中有了其订阅的消息就会触发该方法的执行
// 其返回值为当前consumer消费的状态
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
// 逐条消息消息
for (MessageExt msg : msgs) {
System.out.println(msg);
}
// 返回消息状态，消费成功
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
// 开启消费者消费
consumer.start();
}
四、顺序消息
public static void main(String[] args) throws Exception {
// 创建一个producer，参数为Producer group ID
DefaultMQProducer producer = new DefaultMQProducer("money");
// 指定nameServer地址
producer.setNamesrvAddr("192.168.200.132:9876");
// 开启生产者
producer.start();
// 生产并发送100条消息
for (int i = 0; i < 100; i++) {
Integer orderId = i;
byte[] body = ("money" + i).getBytes();
Message message = new Message("moneyTopic", "moneyTag", body);
// 将orderId作为消息key
message.setKeys(orderId.toString());
// send()的第三个参数值会传递给选择器的select()的第三个参数
// 该send为同步发送
SendResult sendResult = producer.send(message, new MessageQueueSelector() {
// 具体的选择算法在该方法中定义
@Override
public MessageQueue select(List<MessageQueue> mqs, Message message, Object arg) {
// 以下是使用消息key作为 选择的选择算法
String keys = message.getKeys();
Integer id = Integer.valueOf(keys);
// 以下是使用arg作为选择key的选择算法
// Integer id = (Integer) arg;
int index = id % mqs.size();
return mqs.get(index);
}
}, orderId);
System.out.println(sendResult);
}
producer.shutdown();
}
五、延迟消费
public class DelayProducer extends Exception {
public static void main(String[] args) throws Exception {
// 创建一个producer，参数为Producer group ID
DefaultMQProducer producer = new DefaultMQProducer("money");
// 指定nameServer地址
producer.setNamesrvAddr("192.168.200.132:9876");
// 开启生产者
producer.start();
// 生产并发送100条消息
for (int i = 0; i < 10; i++) {
byte[] body = ("money" + i).getBytes();
Message message = new Message("delayTopic", "delayTag", body);
// 指定消息延迟等级为3级，即延迟10s
message.setDelayTimeLevel(3);
SendResult sendResult = producer.send(message);
// 输出消息被发送的时间
System.out.print(new SimpleDateFormat("mm:ss ").format(new Date()));
System.out.println(sendResult);
}
producer.shutdown();
}
}
六、发送事务消息
public class TransactionProducer {
public static void main(String[] args) throws Exception {
// 创建一个producer，参数为Producer group ID
TransactionMQProducer producer = new TransactionMQProducer("money");
// 指定nameServer地址
producer.setNamesrvAddr("192.168.200.132:9876");
/**
* 定义一个线程池
* corePoolSize:线程池中核心线程数量
* maximumPoolSize:线程池中最多线程数
* keepAliveTime:当线程池中线程数量大于核心线程数量时，多余空闲线程的存活时长
* unit:时间单位
* workQueue:临时存放任务的队列，其参数就是队列的长度
* threadFactory:线程工厂
*/
ThreadPoolExecutor executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS,
new ArrayBlockingQueue<Runnable>(2000),
new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
Thread thread = new Thread(r);
thread.setName("client-transation-msg-check-thread");
return thread;
}
});
// 为生产者指定一个线程池
producer.setExecutorService(executorService);
// 为生产者添加事务监听器
producer.setTransactionListener(new ICBCTransactionListener());
producer.start();
String[] tags = {"firstTag", "secondTag", "thirdTag"};
for (int i = 0; i < 3; i++) {
byte[] body = ("moeny" + i).getBytes();
Message message = new Message("transactionTopic", tags[i], body);
// 发送事务消息
// 第二个参数用于指定在本地事务时要使用的业务参数
SendResult sendResult = producer.sendMessageInTransaction(message, null);
System.out.println("发送结果为：" + sendResult);
}
}
}
class ICBCTransactionListener implements TransactionListener {
// 回调操作方法
// 消息预提交成功就会触发该方法的执行，用于完成本地事务
@Override
public LocalTransactionState executeLocalTransaction(Message msg,
Object arg) {
System.out.println("预提交消息成功：" + msg);
// 假设接收到TAGA的消息就表示扣款操作成功，TAGB的消息表示扣款失败，
// TAGC表示扣款结果不清楚，需要执行消息回查
if (StringUtils.equals("firstTag", msg.getTags())) {
return LocalTransactionState.COMMIT_MESSAGE;
} else if (StringUtils.equals("secondTag", msg.getTags())) {
return LocalTransactionState.ROLLBACK_MESSAGE;
} else if (StringUtils.equals("thirdTag", msg.getTags())) {
return LocalTransactionState.UNKNOW;
}
return LocalTransactionState.UNKNOW;
}
// 消息回查方法
// 引发消息回查的原因最常见的有两个：
// 1)回调操作返回UNKNWON
// 2)TC没有接收到TM的最终全局事务确认指令
@Override
public LocalTransactionState checkLocalTransaction(MessageExt msg) {
System.out.println("执行消息回查" + msg.getTags());
return LocalTransactionState.COMMIT_MESSAGE;
}
}
网友评论