做网站 360的好不好,小说排行榜,二次开发软件,网站环境搭建好后怎么做网站大纲 创建RabbitMQ队列新建工程新增依赖编码设置数据源配置读取、处理数据完整代码 打包、上传和运行任务测试 工程代码 在《Java版Flink使用指南——安装Flink和使用IntelliJ制作任务包》一文中#xff0c;我们完成了第一个小型Demo的编写。例子中的数据是代码预先指定的。而… 大纲 创建RabbitMQ队列新建工程新增依赖编码设置数据源配置读取、处理数据完整代码 打包、上传和运行任务测试 工程代码 在《Java版Flink使用指南——安装Flink和使用IntelliJ制作任务包》一文中我们完成了第一个小型Demo的编写。例子中的数据是代码预先指定的。而现实中数据往往来源于外部。本文我们将尝试Flink从RabbitMQ中读取数据然后输出到日志中。 关于RabbitMQ的知识可以参阅《RabbitMQ实践》。
创建RabbitMQ队列
我们创建一个Classic队列data.from.rbtmq。注意要选择Durable类型这是后续用的默认连接器的限制。 具体方法见《RabbitMQ实践——在管理后台测试消息收发功能》。
后续我们将在后台通过默认交换器给这个队列新增消息。
新建工程
我们在IntelliJ中新建一个工程DataFromRabbitMQ。 Archetype填入org.apache.flink:flink-quickstart-java。 版本填入与Flink的版本1.19.1
新增依赖
在pom.xml中新增RabbitMQ连接器
dependencygroupIdorg.apache.flink/groupIdartifactIdflink-connector-rabbitmq/artifactIdversion3.0.1-1.17/version
/dependency编码
设置数据源配置
String queueName data.from.rbtmq;
String host 172.21.112.140; // IP of the rabbitmq server
int port 5672;
String username admin;
String password fangliang;
String virtualHost /;
int parallelism 1;// create a RabbitMQ source
RMQConnectionConfig rmqConnectionConfig new RMQConnectionConfig.Builder().setHost(host).setPort(port).setUserName(username).setPassword(password).setVirtualHost(virtualHost).build();RMQSourceString rmqSource new RMQSource(rmqConnectionConfig, queueName, true, new SimpleStringSchema());读取、处理数据
下面代码通过addSource添加RabbitMQ数据源。注意不能使用fromSource方法是因为RMQSource没有实现SourceFunction方法。
final DataStreamString stream env.addSource(rmqSource).name(username s source from queueName).setParallelism(parallelism);stream.print().name(username s data from queueName);完整代码
/** Licensed to the Apache Software Foundation (ASF) under one* or more contributor license agreements. See the NOTICE file* distributed with this work for additional information* regarding copyright ownership. The ASF licenses this file* to you under the Apache License, Version 2.0 (the* License); you may not use this file except in compliance* with the License. You may obtain a copy of the License at** http://www.apache.org/licenses/LICENSE-2.0** Unless required by applicable law or agreed to in writing, software* distributed under the License is distributed on an AS IS BASIS,* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.* See the License for the specific language governing permissions and* limitations under the License.*/package org.example;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;/*** Skeleton for a Flink DataStream Job.** pFor a tutorial how to write a Flink application, check the* tutorials and examples on the a hrefhttps://flink.apache.orgFlink Website/a.** pTo package your application into a JAR file for execution, run* mvn clean package on the command line.** pIf you change the name of the main class (with the public static void main(String[] args))* method, change the respective entry in the POM.xml file (simply search for mainClass).*/
public class DataStreamJob {public static void main(String[] args) throws Exception {// Sets up the execution environment, which is the main entry point// to building Flink applications.final StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();String queueName data.from.rbtmq;String host 172.21.112.140; // IP of the rabbitmq serverint port 5672;String username admin;String password fangliang;String virtualHost /;int parallelism 1;// create a RabbitMQ sourceRMQConnectionConfig rmqConnectionConfig new RMQConnectionConfig.Builder().setHost(host).setPort(port).setUserName(username).setPassword(password).setVirtualHost(virtualHost).build();RMQSourceString rmqSource new RMQSource(rmqConnectionConfig, queueName, true, new SimpleStringSchema());final DataStreamString stream env.addSource(rmqSource).name(username s source from queueName).setParallelism(parallelism);stream.print().name(username s data from queueName);env.execute(Flink Java API Skeleton);}
}
打包、上传和运行任务 测试
在RabbitMQ后台的默认交换器中发布一条消息到data.from.rbtmq 然后使用下面指令可以看到Flink读取到消息并执行了print方法
tail log/flink-*-taskexecutor-*.outflink-fangliang-taskexecutor-0-fangliang.out data from http://172.21.112.140:15672/#/exchanges/%2F/amq.default 工程代码
https://github.com/f304646673/FlinkDemo
