RocketMQ实现数据分发

爱被打了一巴掌 2022-12-10 15:54 99阅读 0赞

#### 1.生产者 ####

依赖

<dependency>
                <groupId>org.apache.rocketmq</groupId>
                <artifactId>rocketmq-client</artifactId>
                <version>4.7.1</version>
            </dependency>

代码

import org.apache.rocketmq.client.exception.MQBrokerException;
    import org.apache.rocketmq.client.exception.MQClientException;
    import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.common.message.Message;
    import org.apache.rocketmq.remoting.common.RemotingHelper;
    import org.apache.rocketmq.remoting.exception.RemotingException;
    import java.io.UnsupportedEncodingException;
    /** * @description: RocketListener * @author: xxx * @date: 2020-08-05 11:09 */
    public class SyncProducer { 
        public static void main(String[] args) throws Exception { 
            //Instantiate with a producer group name.
            DefaultMQProducer producer = new
                    DefaultMQProducer("qes");
            // Specify name server addresses.
            producer.setNamesrvAddr("127.127.127.127:9876");
            //Launch the instance.
            producer.start();
            for (int i = 0; i < 1; i++) { 
                //Create a message instance, specifying topic, tag and message body.
                try { 
                    String messageJson = "{\"action\":3,\"subject\":\"dim_product\",\"data\":[{\"id\":\"459\",\"brand\":\"Lenovo\",\"productline\":\"PH\",\"productfamily\":\"Lenovo A Series\",\"marketname\":\"Lenovo A7\",\"internalname\":\"AK47\",\"sci_npi_pmd_program_name\":\"N/A\",\"design_type\":\"ODM-Fast Track\",\"odm\":\"Sprocomm\",\"modelname\":\"L19111\",\"blurname\":\"\",\"carrier\":\"N/A\",\"apccode\":\"N/A\",\"sadate\":\"\",\"dualsim\":\"\",\"isbarcode\":\"0\",\"launch_year\":\"FY20\",\"archive\":\"1\",\"design_type0\":\"ODM\",\"updatetime\":\"2020-06-05 17:40:494\"}]}";
                    Message msg = new Message("qesTT" /* Topic */,
                            "product-addOrUpdate" /* Tag */,
                            "OrderId188" /* key */,
                            messageJson.getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body,DEFAULT_CAHRSET:UTF-8*/
                    );
                    //Call send message to deliver message to one of brokers.
                    SendResult sendResult = producer.send(msg);
                    System.out.printf("%s%n", sendResult);
                } catch (UnsupportedEncodingException e) { 
                    e.printStackTrace();
                } catch (MQClientException e) { 
                    e.printStackTrace();
                } catch (RemotingException e) { 
                    e.printStackTrace();
                } catch (MQBrokerException e) { 
                    e.printStackTrace();
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
            //Shut down once the producer instance is not longer in use.
            producer.shutdown();
    
        }
    }

#### 2.消费者 ####

依赖

<dependency>
                <groupId>org.apache.rocketmq</groupId>
                <artifactId>rocketmq-spring-boot</artifactId>
                <version>2.1.1</version>
            </dependency>

消费者配置

rocketmq:
      name-server: 127.127.127.127:9876

代码

import com.alibaba.fastjson.JSON;
    import com.alibaba.fastjson.JSONArray;
    import com.alibaba.fastjson.JSONObject;
    import com.lenovo.qit.system.service.IDimProductService;
    import org.apache.commons.collections.CollectionUtils;
    import org.apache.commons.lang3.StringUtils;
    import org.apache.rocketmq.spring.annotation.ConsumeMode;
    import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
    import org.apache.rocketmq.spring.core.RocketMQListener;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.stereotype.Component;
    
    import javax.annotation.Resource;
    import java.util.Arrays;
    import java.util.List;
    
    /** * @description: RocketListener * @author: xxx * @date: 2020-08-05 11:09 */
    @RocketMQMessageListener(topic = "qes",
            consumerGroup = "qit",
            selectorExpression = "qit || sqm|qit",
            consumeMode = ConsumeMode.ORDERLY)
    @Component
    public class RocketListener
            implements RocketMQListener<String>
    {

参考：  
[https://blog.csdn.net/zhoudingding/article/details/106942574][https_blog.csdn.net_zhoudingding_article_details_106942574]

[https_blog.csdn.net_zhoudingding_article_details_106942574]: https://blog.csdn.net/zhoudingding/article/details/106942574