Flink-Connectors(连接器)（3）RabbitMQ_flink rabbitmq

flink 提供了专门操作RabbitMQ的连接器，使用起来更方便，配置连接信息即可快速实现数据读取与输出，但目前仅支持Queue模式，如需使用交换机模式，仍需要自定义RabbitMQ 数据源读取与数据

必要依赖

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-rabbitmq_2.12</artifactId>
    <version>1.12.2</version>
</dependency>
1
2
3
4
5

代码示例

package com.leilei;

import com.alibaba.fastjson.JSON;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.rabbitmq.RMQSink;
import org.apache.flink.streaming.connectors.rabbitmq.RMQSource;
import org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig;

import java.nio.charset.StandardCharsets;

/**
 * @author lei
 * @version 1.0
 * @date 2021/3/14 15:27
 * @desc flink 连接器 rabbitmq
 */
public class FlinkConnectorsRabbitMq {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        final RMQConnectionConfig rabbitConfig = new RMQConnectionConfig.Builder()
                .setHost("xxx")
                .setUserName("admin")
                .setPassword("xx")
                .setPort(5672)
                .setVirtualHost("/")
                .build();
        //rabbit connectors 加载数据
        DataStream<String> stream = env
                .addSource(new RMQSource<>(
                        // mq 连接配置
                        rabbitConfig,
                        // 队列名
                        "vehicle-location",
                        true,
                        // 反序列化方式
                        new SimpleStringSchema(StandardCharsets.UTF_8)))
                .setParallelism(1);
        stream.print();
        //rabbit connectors 数据输出
        stream.addSink(new RMQSink<>(
                rabbitConfig,
                // 输出到哪个队列
                "over-speeding-alarm",
                // 序列化方式
                new SimpleStringSchema(StandardCharsets.UTF_8)));
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class Location {
        private String id;
        private String licensePlate;
        private String plateColor;
        private Integer speed;
        private Integer limitSpeed;
        private Long deviceTime;
        private String zone;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77