Rocketmq4.7 生产者、push消费者和pull消费者 java客户端简单使用和一些注意事项
内容导读
互联网集市收集整理的这篇技术教程文章主要介绍了Rocketmq4.7 生产者、push消费者和pull消费者 java客户端简单使用和一些注意事项，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含6260字，纯文字阅读大概需要9分钟。
内容图文
maven引用
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>4.7.0</version>
</dependency>
<!--一个好用的工具包，可以不引入-->
<dependency>
<groupId>cn.hutool</groupId>
<artifactId>hutool-all</artifactId>
<version>5.3.0</version>
</dependency>
说明
不管是生产者还是消费者，都有很多参数可以配置，rocketmq命名比较好，基本可以从参数名上判断具体作用，还有注释可以看。
下面例子中只给出了常用的一些参数设置。更多参数可自行探索。
简单生产者实现
注意:
1、NamesrvAddr参数在多个节点时，用英文分号分隔，例: 192.168.9.58:9876;192.168.9.59:9876
import cn.hutool.core.util.RandomUtil;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
public class Producer {
public static void main(String[] args) throws Exception {
DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
producer.setNamesrvAddr("127.0.0.1:9876");
//发送超时时间，默认3000 单位ms
producer.setSendMsgTimeout(5000);
producer.start();
try {
Message msg = new Message("TestTopic",// topic
"177", // tag 可以为空，用以简单的筛选。
RandomUtil.randomString(8), // key 可以为空，可用以查询。
("test" + RandomUtil.randomString(8)).getBytes()); // body ，我常将对象转json再获取byte[] 进行传输。
SendResult send = producer.send(msg);
if (send.getSendStatus().equals(SendStatus.SEND_OK)) {
//发送成功处理
}else {
//发送失败处理
}
} catch (Exception e) {
//发送失败处理
e.printStackTrace();
}
//正式环境不要发完就就shutdown，要在应用退出时再shutdown。
producer.shutdown();
}
}
多线程加批量生产者模拟实现
注意:
1、批量发送时，topic必须为同一个，否则发送会报异常。
2、批量发送相较于单条发送速度提升很大。
import cn.hutool.core.util.RandomUtil;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
public class Producer {
public static void main(String[] args) throws Exception {
DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
producer.setNamesrvAddr("127.0.0.1:9876");
//发送超时时间，默认3000 单位ms
producer.setSendMsgTimeout(5000);
producer.start();
int threadCount = 20;
int forCount = 100000;
CountDownLatch latch = new CountDownLatch(threadCount);
long start = System.currentTimeMillis();
for (int i = 0; i < threadCount; i++) {
new Thread(() -> {
try {
List<Message> list = new ArrayList<>();
for (int j = 0; j < forCount; j++) {
try {
Message msg = new Message("TestTopic",// topic
"177", // tag
RandomUtil.randomString(8), // key
("test" + RandomUtil.randomString(8)).getBytes()); // body
list.add(msg);
if (list.size() >= 100) {
SendResult send = producer.send(list);
if (send.getSendStatus().equals(SendStatus.SEND_OK)) {
//发送成功处理
list.clear();
}else {
//发送失败处理
}
}
} catch (Exception e) {
//发送失败处理
e.printStackTrace();
}
}
if (list.size() > 0) {
SendResult send = producer.send(list);
if (!send.getSendStatus().equals(SendStatus.SEND_OK)) {
System.out.println(send);
}
list.clear();
}
} catch (Exception e) {
e.printStackTrace();
} finally {
latch.countDown();
}
}).start();
}
latch.await();
long hs = System.currentTimeMillis() - start;
System.out.println(hs);
long speed = (threadCount * forCount) / (hs >= 0 ? 1 : hs / 1000);
System.out.println("速度" + speed);
//正式环境不要发完就就shutdown，要在应用退出时再shutdown。
producer.shutdown();
}
}
push消费者
import cn.hutool.core.lang.Console;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
public class PushConsumer {
public static void main(String[] args) throws Exception {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("PushConsumerGroupName");
consumer.setNamesrvAddr("127.0.0.1:9876");
//一个GroupName第一次消费时的位置
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
consumer.setConsumeThreadMin(20);
consumer.setConsumeThreadMax(20);
//要消费的topic，可使用tag进行简单过滤
consumer.subscribe("TestTopic", "*");
//一次最大消费的条数
consumer.setConsumeMessageBatchMaxSize(100);
//消费模式，广播或者集群，默认集群。
consumer.setMessageModel(MessageModel.CLUSTERING);
//在同一jvm中 需要启动两个同一GroupName的情况需要这个参数不一样。
consumer.setInstanceName("InstanceName");
//配置消息监听
consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
try {
//业务处理
msgs.forEach(msg -> {
Console.log(msg);
});
} catch (Exception e) {
System.err.println("接收异常" + e);
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});
consumer.start();
System.out.println("Consumer Started.");
}
}
pull消费者
从4.6之后，提供了DefaultLitePullConsumer 大大简化了pull的操作。以下为新实现，4.6之前的版本不支持。
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.lang.Console;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;
public class PullConsumer {
private static boolean runFlag = true;
public static void main(String[] args) throws Exception {
DefaultLitePullConsumer consumer = new DefaultLitePullConsumer("PullConsumerGroupName");
consumer.setNamesrvAddr("127.0.0.1:9876");
//要消费的topic，可使用tag进行简单过滤
consumer.subscribe("TestTopic", "*");
//一次最大消费的条数
consumer.setPullBatchSize(100);
//无消息时，最大阻塞时间。默认5000 单位ms
consumer.setPollTimeoutMillis(5000);
consumer.start();
while (runFlag){
try {
//拉取消息，无消息时会阻塞
List<MessageExt> msgs = consumer.poll();
if (CollUtil.isEmpty(msgs)){
continue;
}
//业务处理
msgs.forEach(msg-> Console.log(new String(msg.getBody())));
//同步消费位置。不执行该方法，应用重启会存在重复消费。
consumer.commitSync();
}catch (Exception e){
e.printStackTrace();
}
}
consumer.shutdown();
}
}
内容总结
以上是互联网集市为您收集整理的Rocketmq4.7 生产者、push消费者和pull消费者 java客户端简单使用和一些注意事项全部内容，希望文章能够帮你解决Rocketmq4.7 生产者、push消费者和pull消费者 java客户端简单使用和一些注意事项所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。
内容备注
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。
内容手机端
扫描二维码推送至手机访问。
来源：【匿名】