springboot集成RocketMQ,三种方式(原生Jar,springboot封装starter,阿里云Ons接入) |
您所在的位置：网站首页 › 数据整合的三种方式包括 › springboot集成RocketMQ,三种方式(原生Jar,springboot封装starter,阿里云Ons接入) |
|
写在前面
这里介绍下Springboot 集成RocketMQ的三种方式 一、原生 jar（rocketmq-client） 1.1、producer 1.1.1、三个基本使用 producerGroup，定义生产者组 DefaultMQProducer，定义生产者配置 TransactionMQProducer，定义支持事务生产者 1.1.2、三种基本发送方式： 同步发送 异步发送 单项发送同步发送，代码示例 /** * 同步发送实体对象消息 * 可靠同步发送：同步发送是指消息发送方发出数据后，会在收到接收方发回响应之后才发下一个数据包的通讯方式； * 特点：速度快；有结果反馈；数据可靠； * 应用场景：应用场景非常广泛，例如重要通知邮件、报名短信通知、营销短信系统等； * * @param topic * @param tags * @param body * @return * @throws InterruptedException * @throws RemotingException * @throws MQClientException * @throws MQBrokerException * @throws UnsupportedEncodingException */ public String syncSend(String topic, String tags, String body) throws InterruptedException, RemotingException, MQClientException, MQBrokerException, UnsupportedEncodingException { Message message = new Message(topic, tags, body.getBytes(RemotingHelper.DEFAULT_CHARSET)); Message msg = new Message(topic /* Topic */, tags /* Tag */, ("Hello RocketMQ ").getBytes() /* Message body */ ); // 发送消息到一个Broker SendResult sendResult = producer.send(msg); // 通过sendResult返回消息是否成功送达 System.out.printf("%s%n", sendResult); TimeUnit.SECONDS.sleep(1); return "{\"MsgId\":\"" + sendResult.getMsgId() + "\"}"; }异步发送，代码示例 /** * 异步发送消息 * 可靠异步发送：发送方发出数据后，不等接收方发回响应，接着发送下个数据包的通讯方式； * 特点：速度快；有结果反馈；数据可靠； * 应用场景：异步发送一般用于链路耗时较长,对 rt响应时间较为敏感的业务场景,例如用户视频上传后通知启动转码服务,转码完成后通知推送转码结果等； * * @param topic * @param tags * @param body * @return * @throws InterruptedException * @throws RemotingException * @throws MQClientException * @throws MQBrokerException * @throws UnsupportedEncodingException */ public void asyncSend(String topic, String tags, String body) throws Exception { Message msg = new Message(topic /* Topic */, tags /* Tag */, ("Hello RocketMQ ").getBytes() /* Message body */ ); // 发送消息到一个Broker producer.send(msg, new SendCallback() { public void onSuccess(SendResult sendResult) { System.out.println("发送结果 ： " + sendResult); } public void onException(Throwable throwable) { System.out.println(throwable.getMessage()); } }); TimeUnit.SECONDS.sleep(1); }单项发送，代码示例 /** * 单向发送 * 单向发送：只负责发送消息，不等待服务器回应且没有回调函数触发，即只发送请求不等待应答；此方式发送消息的过程耗时非常短，一般在微秒级别； * 特点：速度最快，耗时非常短，毫秒级别；无结果反馈；数据不可靠，可能会丢失； * 应用场景：适用于某些耗时非常短，但对可靠性要求并不高的场景，例如日志收集； * * @param topic * @param tags * @param body * @throws InterruptedException * @throws RemotingException * @throws MQClientException * @throws MQBrokerException * @throws UnsupportedEncodingException */ public void oneway(String topic, String tags, String body) throws Exception { Message msg = new Message(topic /* Topic */, tags /* Tag */, ("Hello RocketMQ ").getBytes() /* Message body */ ); producer.sendOneway(msg); TimeUnit.SECONDS.sleep(1); } 1.1.3、其他发送特性 消息延迟 设置消息属性，用于消费过滤 消息队列选择器Selector 事务监听消息延迟 /** * 延迟 消费 */ public void delayTestListener() { DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("delayGroup"); consumer.setNamesrvAddr(namesrvAddr); try { // 订阅PushTopic下Tag为push的消息,都订阅消息 consumer.subscribe("delayPushMsg", "push"); // 程序第一次启动从消息队列头获取数据 consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET); //可以修改每次消费消息的数量，默认设置是每次消费一条 consumer.setConsumeMessageBatchMaxSize(1); //在此监听中消费信息，并返回消费的状态信息 consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> { // 会把不同的消息分别放置到不同的队列中 for (MessageExt msg : msgs) { log.info("Receive message:msgId={},msgBody={},delay={} ms", msg.getMsgId(), new String(msg.getBody()), (System.currentTimeMillis() - msg.getStoreTimestamp())); } return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; }); consumer.start(); } catch (Exception e) { e.printStackTrace(); } }设置消息属性，用于消费过滤 /** * todo 注意这里 需要启动 broker 前，设置 支持SQL92 Filter = enable * sql filter * * @param topic * @param tags * @param body * @param i * @throws Exception */ public void filtersql(String topic, String tags, String body, int i) throws Exception { //消息 Message message = new Message(topic, tags, body.getBytes()) |
今日新闻 |
推荐新闻 |
| CopyRight 2018-2019 办公设备维修网 版权所有 豫ICP备15022753号-3 |