How to connect Connectors to RabbitMq in Flink
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!
