Shulou(Shulou.com)06/01 Report--

This article is about how Connectors connects to RabbitMq in Flink. The editor thinks it is very practical, so share it with you as a reference and follow the editor to have a look.

Connect to RabbitMq message queuing middleware by using Flink DataStream Connectors data flow connector and provide data flow input and output operations

Sample environment

Java.version: 1.8.xflink.version: 1.11.1rabbitMq:3.5.7

Sample data source (project code cloud download)

Building Development Environment and data of Flink system example

Sample module (pom.xml)

DataStream Connectors and sample Module of Flink system

Data stream input

DataStreamSource.java

Package com.flink.examples.rabbitmq;import org.apache.flink.api.common.serialization.SimpleStringSchema;import org.apache.flink.streaming.api.datastream.DataStream;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 / * * @ Description takes data from MQ and outputs it to DataStream stream * / public class DataStreamSource {/ * official document: https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/dev/connectors/rabbitmq.html * / public static void main (String [] args) throws Exception {final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment () RMQConnectionConfig connectionConfig = new RMQConnectionConfig.Builder () .setHost ("127.0.0.1") .setPort (5672) .setUserName ("admin") .setPassword ("admin") .setVirtualHost ("datastream") .build () Final DataStream stream = env .addSource (new RMQSource (connectionConfig, "test", true, new SimpleStringSchema ()) .setParallelism (1); stream.print (); env.execute ("flink rabbitMq source");}}

Data stream output

DataStreamSink.java

Package com.flink.examples.rabbitmq;import org.apache.flink.api.common.serialization.SimpleStringSchema;import org.apache.flink.streaming.api.datastream.DataStream;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.streaming.connectors.rabbitmq.RMQSink;import org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig / * * @ Description outputs the data from the DataStream stream to the rabbitMq queue * / public class DataStreamSink {/ * official document: https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/dev/connectors/rabbitmq.html * / public static void main (String [] args) throws Exception {final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment () RMQConnectionConfig connectionConfig = new RMQConnectionConfig.Builder () .setHost ("127.0.0.1") .setPort (5672) .setUserName ("admin") .setPassword ("admin") .setVirtualHost ("datastream") .build () String [] words = new String [] {"props", "student", "build", "name", "execute"}; final DataStream stream = env.fromElements (words); stream.addSink (new RMQSink (connectionConfig, "test", new SimpleStringSchema ()); env.execute ("flink rabbitMq sink");}}

Data presentation

Thank you for reading! This is the end of the article on "how to connect Connectors to RabbitMq in Flink". I hope the above content can be of some help to you, so that you can learn more knowledge. if you think the article is good, you can share it out for more people to see!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.