2. 程式人生 > 實用技巧 >Flink 從 0 到 1 學習之（13）Flink 讀取 Kafka 資料寫入到 RabbitMQ FlinkKafkaRabbitMQ大資料流式計算

Flink 從 0 到 1 學習之（13）Flink 讀取 Kafka 資料寫入到 RabbitMQ FlinkKafkaRabbitMQ大資料流式計算

阿新 • • 發佈：2020-09-07

前言

之前有文章 《從0到1學習Flink》—— Flink 寫入資料到 Kafka寫過 Flink 將處理後的資料後發到 Kafka 訊息佇列中去，當然我們常用的訊息佇列可不止這一種，還有 RocketMQ、RabbitMQ 等，剛好 Flink 也支援將資料寫入到 RabbitMQ，所以今天我們就來寫篇文章講講如何將 Flink 處理後的資料寫入到 RabbitMQ。

前提準備

安裝 RabbitMQ

這裡我直接用 docker 命令安裝吧，先把 docker 在 mac 上啟動起來。

在命令列中執行下面的命令：

1
 
docker run -d  -p 15672:15672  -p  5672:5672  -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin --name rabbitmq rabbitmq:3-management

對這個命令不懂的童鞋可以看看我以前的文章：http://www.54tianzhisheng.cn/2018/01/26/SpringBoot-RabbitMQ/

登入使用者名稱和密碼分別是：admin / admin ，登入進去是這個樣子就代表安裝成功了：

依賴

pom.xml 中新增 Flink connector rabbitmq 的依賴如下：

1
2
3
4
5
 
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-rabbitmq_${scala.binary.version}</artifactId
 
>
    <version>${flink.version}</version>
</dependency>

生產者

這裡我們依舊自己寫一個工具類一直的往 RabbitMQ 中的某個 queue 中發資料，然後由 Flink 去消費這些資料。

注意按照我的步驟來一步步操作，否則可能會出現一些錯誤！

RabbitMQProducerUtil.java

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
 
import com.rabbitmq.client.Channel;
 

import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class RabbitMQProducerUtil {
    public final static String QUEUE_NAME = "zhisheng";

    public static void main(String[] args) throws Exception {
        //建立連線工廠
        ConnectionFactory factory = new ConnectionFactory();

        //設定RabbitMQ相關資訊
        factory.setHost("localhost");
        factory.setUsername("admin");
        factory.setPassword("admin");
        factory.setPort(5672);

        //建立一個新的連線
        Connection connection = factory.newConnection();

        //建立一個通道
        Channel channel = connection.createChannel();

        // 宣告一個佇列
//        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        //傳送訊息到佇列中
        String message = "Hello zhisheng";

        //我們這裡演示傳送一千條資料
        for (int i = 0; i < 1000; i++) {
            channel.basicPublish("", QUEUE_NAME, null, (message + i).getBytes("UTF-8"));
            System.out.println("Producer Send +'" + message + i);
        }

        //關閉通道和連線
        channel.close();
        connection.close();
    }
}
 
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
 
import com.zhisheng.common.utils.ExecutionEnvUtil;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.rabbitmq.RMQSource;
import org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig;

/**
 * 從 rabbitmq 讀取資料
 */
public class Main {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        ParameterTool parameterTool = ExecutionEnvUtil.PARAMETER_TOOL;

        //這些配置建議可以放在配置檔案中，然後通過 parameterTool 來獲取對應的引數值
        final RMQConnectionConfig connectionConfig = new RMQConnectionConfig
                .Builder().setHost("localhost").setVirtualHost("/")
                .setPort(5672).setUserName("admin").setPassword("admin")
                .build();

        DataStreamSource<String> zhisheng = env.addSource(new RMQSource<>(connectionConfig,
                "zhisheng",
                true,
                new SimpleStringSchema()))
                .setParallelism(1);
        zhisheng.print();

        //如果想保證 exactly-once 或 at-least-once 需要把 checkpoint 開啟
//        env.enableCheckpointing(10000);
        env.execute("flink learning connectors rabbitmq");
    }
}

執行 RabbitMQProducerUtil 類，再執行 Main 類！

注意⚠️：

1、RMQConnectionConfig 中設定的使用者名稱和密碼要設定成 admin/admin，如果你換成是 guest/guest，其實是在 RabbitMQ 裡面是沒有這個使用者名稱和密碼的，所以就會報這個錯誤：

1
 
nested exception is com.rabbitmq.client.AuthenticationFailureException: ACCESS_REFUSED - Login was refused using authentication mechanism PLAIN. For details see the broker logfile.

不出意外的話應該你執行 RabbitMQProducerUtil 類後，立馬兩個執行的結果都會出來，速度還是很快的。

2、如果你在 RabbitMQProducerUtil 工具類中把註釋的那行程式碼開啟的話：

1
2
 
 // 宣告一個佇列
//        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

就會出現這種錯誤：

1
 
Caused by: com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - inequivalent arg 'durable' for queue 'zhisheng' in vhost '/': received 'true' but current is 'false', class-id=50, method-id=10)

這是因為你開啟那個註釋的話，一旦你運行了該類就會建立一個叫做 zhisheng 的 Queue，當你再執行 Main 類中的時候，它又會建立這樣一個叫 zhisheng 的 Queue，然後因為已經有同名的 Queue 了，所以就有了衝突，解決方法就是把那行程式碼註釋就好了。

3、該 connector（聯結器）中提供了 RMQSource 類去消費 RabbitMQ queue 中的訊息和確認 checkpoints 上的訊息，它提供了三種不一樣的保證：

  • Exactly-once(只消費一次): 前提條件有，1 是要開啟 checkpoint，因為只有在 checkpoint 完成後，才會返回確認訊息給 RabbitMQ（這時，訊息才會在 RabbitMQ 佇列中刪除)；2 是要使用 Correlation ID，在將訊息發往 RabbitMQ 時，必須在訊息屬性中設定 Correlation ID。資料來源根據 Correlation ID 把從 checkpoint 恢復的資料進行去重；3 是資料來源不能並行，這種限制主要是由於 RabbitMQ 將訊息從單個佇列分派給多個消費者。
  • At-least-once(至少消費一次): 開啟了 checkpoint，但未使用相 Correlation ID 或 資料來源是並行的時候，那麼就只能保證資料至少消費一次了
  • No guarantees(無法保證): Flink 接收到資料就返回確認訊息給 RabbitMQ

Sink 資料到 RabbitMQ

RabbitMQ 除了可以作為資料來源，也可以當作下游，Flink 消費資料做了一些處理之後也能把資料發往 RabbitMQ，下面演示下 Flink 消費 Kafka 資料後寫入到 RabbitMQ。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
 
public class Main1 {
    public static void main(String[] args) throws Exception {
        final ParameterTool parameterTool = ExecutionEnvUtil.createParameterTool(args);
        StreamExecutionEnvironment env = ExecutionEnvUtil.prepare(parameterTool);
        DataStreamSource<Metrics> data = KafkaConfigUtil.buildSource(env);

        final RMQConnectionConfig connectionConfig = new RMQConnectionConfig
                .Builder().setHost("localhost").setVirtualHost("/")
                .setPort(5672).setUserName("admin").setPassword("admin")
                .build();

        //注意，換一個新的 queue，否則也會報錯
        data.addSink(new RMQSink<>(connectionConfig, "zhisheng001", new MetricSchema()));
        env.execute("flink learning connectors rabbitmq");
    }
}

是不是很簡單？但是需要注意的是，要換一個之前不存在的 queue，否則是會報錯的。

不出意外的話，你可以看到 RabbitMQ 的監控頁面會出現新的一個 queue 出來，如下圖：

總結

本文先把 RabbitMQ 作為資料來源，寫了個 Flink 消費 RabbitMQ 佇列裡面的資料進行打印出來，然後又寫了個 Flink 消費 Kafka 資料後寫入到 RabbitMQ 的例子！

相關推薦

Flink 從 0 到 1 學習之（13）Flink 讀取 Kafka 資料寫入到 RabbitMQ FlinkKafkaRabbitMQ大資料流式計算

前言 之前有文章 《從0到1學習Flink》—— Flink 寫入資料到 Kafka寫過 Flink 將處理後的資料後發到 Kafka 訊息佇列中去，當然我們常用的訊息佇列可不止這一種，還有 RocketMQ、RabbitMQ 等，剛好 Flink

Flink 從 0 到 1 學習之（2） Flink 配置檔案詳解

前面文章我們已經知道 Flink 是什麼東西了，安裝好 Flink 後，我們再來看下安裝路徑下的配置檔案吧。

Flink 從 0 到 1 學習之（10）Flink 專案如何執行？

前言 之前寫了不少 Flink 文章了，也有不少 demo，但是文章寫的時候都是在本地直接執行 Main 類的 main 方法，其實 Flink 是支援在 UI 上上傳 Flink Job 的 jar 包，然後執行得。

Flink 從 0 到 1 學習之（12）Flink 讀取 Kafka 資料批量寫入到 MySQL

前言 之前其實在 《從0到1學習Flink》—— 如何自定義 Data Sink ？文章中其實已經寫了點將資料寫入到 MySQL，但是一些配置化的東西當時是寫死的，不能夠通用，最近知識星球裡有朋友叫我: 寫個從 kafka 中

Flink 從 0 到 1 學習之（14）Flink 讀取kafka資料，寫入到Hbase

概述 環境說明 scala: 2.12.8 linux下scala安裝部署 flink : 1.8.1Flink1.8.1 叢集部署 kafka_2.12-2.2.0kafka_2.12-2.2.0 叢集部署

Flink 從 0 到 1 學習之（16）Flink DataStream之Kafka資料寫入HDFS，並分割槽到Hive

因業務要求，我們需要從Kafka中讀取資料，變換後最終Sink到業務的訊息佇列中，為保證資料的可靠性，我們同時對Sink的結果資料，進行儲存。最終選擇將流資料Sink到HDFS上，在Flink中，同時也提供了HDFS Connector。下

Flink 從 0 到 1 學習之（18）Flink SQL讀取kafka資料並通過JDBC方式寫入Clickhouse實時場景的簡單例項

說明 讀取kafka資料並且經過ETL後，通過JDBC存入clickhouse中 程式碼 定義POJO類： 12345678

Flink 從 0 到 1 學習之（20）Flink讀取hdfs檔案

接一下以一個示例配置來介紹一下如何以Flink連線HDFS 1. 依賴HDFS pom.xml 新增依賴

Flink 從 0 到 1 學習之（21）Flink讀HBase並寫入HBase

這裡讀HBase提供兩種方式，一種是繼承RichSourceFunction，重寫父類方法，一種是實現OutputFormat介面，具體程式碼如下：

Flink 從 0 到 1 學習之（23）Flink 讀取hive並寫入hive

1，讀取實現了，也是找的資料，核心就是實現了 HCatInputFormatHCatInputFormatBase 上面這兩個類，底層也是 繼承實現了 RichInputFormat：

Flink 從 0 到 1 學習之（25）Flink從redis中獲取資料作為source源

redis中的資料：需要實現SourceFunction介面，指定泛型<>,也就是獲取redis裡的資料，處理完後的資料輸入的資料型別 這裡我們需要的是（我們需要返回kv對的，就要考慮HashMap）pom.xml

Flink 從 0 到 1 學習之（24）Flink將kafka的資料存到redis中

1、依賴 <properties> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>

Flink 從 0 到 1 學習之（26）Flink採集kafka資料後存到mongodb

實體類： import java.io.Serializable; public class FlinkDao implements Serializable{ private String id;

Flink 從 0 到 1 學習之（3） Data Source 介紹

前言 Data Sources 是什麼呢？就字面意思其實就可以知道：資料來源。 Flink 做為一款流式計算框架，它可用來做批處理，即處理靜態的資料集、歷史的資料集；也可以用來做流處理，即實時的處理些實時資料流，實時的產

Flink 從 0 到 1 學習之（5）如何自定義 Data Source ？

前言 我給大家介紹了 Flink Data Source 以及簡短的介紹了一下自定義 Data Source，這篇文章更詳細的介紹下，並寫一個 demo 出來讓大家理解。

Flink 從 0 到 1 學習之（6）如何自定義 Data Sink ？

前言 前篇文章 介紹了 Flink Data Sink，也介紹了 Flink 自帶的 Sink，那麼如何自定義自己的 Sink 呢？這篇文章將寫一個 demo 教大家將從 Kafka Source 的資料 Sink 到 MySQL 中去。

Flink 從 0 到 1 學習之（8）介紹Flink中的Stream Windows

前言 目前有許多資料分析的場景從批處理到流處理的演變， 雖然可以將批處理作為流處理的特殊情況來處理，但是分析無窮集的流資料通常需要思維方式的轉變並且具有其自己的術語（例如，“windowing（視窗化）&rd

SparkStreaming專案實戰 從 0 到 1 學習之（1）使用 Kafka + Spark Streaming + Cassandra 構建資料實時處理引擎

Apache Kafka 是一個可擴充套件，高效能，低延遲的平臺，允許我們像訊息系統一樣讀取和寫入資料。我們可以很容易地在 Java 中使用 Kafka。

Spark專案實戰從0到1之（13）hive經典50題

一.建表和載入資料 1.student表 create table if not exists student(s_id int,s_name string,s_birth string,s_sex string)

Spark專案實戰從0到1之（3）spark讀取hive資料

1.首先將叢集的這3個檔案hive-site.xml，core-size.xml,hdfs-site.xml放到資原始檔裡(必須，否則報錯)