在基于Flink流处理的动态实时(一)的基础上 我写了centos7中安装zookpeer和kafka(单机)的原因是因为不确定公司到底是想用kafka还是rockermq,所以在还没有正式的开发的时候去玩玩kafka。之后决定用rockermq了,所以我接下来写的技术flink和rockermq的一个整合。

首先要了解一下rockermq,初学的话可以看一下初试RocketMQ消息中间件丶一个站在Java后端设计之路的男青年个人博客网站文章,并安装RocketMQ 插件方便省事。

依次运行mqnamesrv.cmd脚本和mqbroker.cmd脚本,因为会出现中文乱码的缘故,我们在rocketmq-console项目的配置文件中添加和配置端口号

server.port=9875
server.tomcat.uri-encoding=UTF-8

并且将namesrvAddr修改成自己的ip地址

rocketmq.config.namesrvAddr=xxx.xxx.xx.xx:9876

然后运行它(随便怎么搞,反正启动就行了。在idea中也行,大jar启动也行)

之后在你游览器输入http://localhost:9875/#/ 可以选择中文还是英文。

在然后在你的项目中加入mq的配置

spring.rocketmq.nameServer=xx.xxx.xx.xxx:9876

在创建一个消费者的一个类,测试是否能发送成功

public class TestDome  {
    /**
     * 消费者如需订阅某Topic下特定类型的消息!
     * @param args
     * @throws InterruptedException
     * @throws MQClientException
     */
    public static void main(String [] args) throws InterruptedException, MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("bk_group");//消费者群组
        consumer.setNamesrvAddr ("xx.xxx.xx.xx:9876");//mq的ip地址和端口
        consumer.setInstanceName(UUID.randomUUID().toString());
        // 请明确标明Tag：只关注自己需要的！
        consumer.subscribe("test", "*");
        consumer. registerMessageListener (new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs ,
                                                            ConsumeConcurrentlyContext context) {
                for (MessageExt messageExt:msgs) {
                    String msg = new String(messageExt.getBody()); //从mq中取得消息
                    System.out.println(Thread.currentThread().getName() + " 接收消息: " + msg);
                }

                MessageExt msg = msgs.get (0);
                /*** 对topic tag验证：只关注特定Pay*/
                if (msg. getTopic (). equals("TopicModel") && msg. getTags (). equals("Pay")) {
                    System.out.print("特定类型:" + new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.out.println("Consumer Started");
    }

上面的消费者群组,你也可以换的。

这是我的主题,你也可以修改的。

修改之后启动他,发送信息

启动你的main方法之后提交信息,先提交启动都可以,无所谓

成功消费,最上面一条是启动直接生产的。

成功搞定,等待下一次flink和rockermq的一个整合,写的不一定特别对,毕竟我也是第一次接触rockermq和flink,如果觉得我写的不对的地方,欢迎留言,互相学习学习。