当前位置: 首页 > news >正文

Java版Flink使用指南——从RabbitMQ中队列中接入消息流

大纲

  • 创建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连接器

<dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-rabbitmq</artifactId><version>3.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();RMQSource<String> rmqSource = new RMQSource<>(rmqConnectionConfig, queueName, true, new SimpleStringSchema());

读取、处理数据

下面代码通过addSource添加RabbitMQ数据源。注意,不能使用fromSource方法,是因为RMQSource没有实现SourceFunction方法。

final DataStream<String> 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.** <p>For a tutorial how to write a Flink application, check the* tutorials and examples on the <a href="https://flink.apache.org">Flink Website</a>.** <p>To package your application into a JAR file for execution, run* 'mvn clean package' on the command line.** <p>If 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();RMQSource<String> rmqSource = new RMQSource<>(rmqConnectionConfig, queueName, true, new SimpleStringSchema());final DataStream<String> 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-*.out

==> flink-fangliang-taskexecutor-0-fangliang.out <==
data from http://172.21.112.140:15672/#/exchanges/%2F/amq.default

工程代码

https://github.com/f304646673/FlinkDemo

相关文章:

Java版Flink使用指南——从RabbitMQ中队列中接入消息流

大纲 创建RabbitMQ队列新建工程新增依赖编码设置数据源配置读取、处理数据完整代码 打包、上传和运行任务测试 工程代码 在《Java版Flink使用指南——安装Flink和使用IntelliJ制作任务包》一文中&#xff0c;我们完成了第一个小型Demo的编写。例子中的数据是代码预先指定的。而…...

Python酷库之旅-第三方库Pandas(013)

目录 一、用法精讲 31、pandas.read_feather函数 31-1、语法 31-2、参数 31-3、功能 31-4、返回值 31-5、说明 31-6、用法 31-6-1、数据准备 31-6-2、代码示例 31-6-3、结果输出 32、pandas.DataFrame.to_feather函数 32-1、语法 32-2、参数 32-3、功能 32-4、…...

Linux 高级 Shell 脚本编程:掌握 Shell 脚本精髓,提升工作效率

【Linux】 高级 Shell 脚本编程&#xff1a;掌握 Shell 脚本精髓&#xff0c;提升工作效率 Shell 脚本编程是 Linux 系统管理员和开发人员的必备技能。通过学习高级 Shell 脚本编程&#xff0c;你可以编写更高效、更灵活和更易于维护的脚本。本文将介绍 Shell 脚本编程中的函数…...

【ARMv8/v9 GIC 系列 1.5 -- Enabling the distribution of interrupts】

请阅读【ARM GICv3/v4 实战学习 】 文章目录 Enabling the distribution of interruptsGIC Distributor 中断组分发控制CPU Interface 中断组分发控制Physical LPIs 的启用Summary Enabling the distribution of interrupts 在ARM GICv3和GICv4体系结构中&#xff0c;中断分发…...

《mysql篇》--索引事务

索引 索引的介绍 索引是帮助MySQL高效获取数据的数据结构&#xff0c;是一种特殊的文件&#xff0c;包含着对数据表里所有记录的引用指针&#xff0c;因为索引本身也比较大&#xff0c;所以索引一般是存储在磁盘上的&#xff0c;索引的种类有很多&#xff0c;不过如果没有特殊…...

科研绘图系列:R语言STAMP图(STAMP Plot)

介绍 STAMP图(STAMP plot)并非一个广泛认知的、具有特定名称的图表类型,而是可能指在STAMP(Statistical Analysis of Metagenomic Profiles:“STAMP: statistical analysis of taxonomic and functional profiles”)软件使用过程中生成的各种统计和可视化图表的总称。ST…...

运维团队如何应对动环监控与IT监控分离的挑战

IT与机房动环监控的一体化是当下及未来的必然趋势&#xff0c;这一模式显著节省了运维过程中的时间与成本。一体化平台不仅消除了频繁切换系统的繁琐&#xff0c;更在一个统一界面上实现了多元化的管理运维功能&#xff0c;极大地提升了工作效率。 在机房升级或新建项目中&…...

深入解析大数据核心概念:数据平台、数据中台、数据湖与数据仓库的异同与应用

大数据领域内的诸多概念常常让人困惑&#xff0c;其中数据平台、数据中台、数据湖和数据仓库是最为关键的几个。 1. 数据平台 定义&#xff1a; 数据平台是一个综合性的技术框架&#xff0c;旨在支持整个数据生命周期的管理和使用。它包含数据采集、存储、处理、分析和可视化…...

开发指南040-业务操作日志

平台所有业务操作都存储在核心库&#xff0c;以便统一分析处理。各业务微服务通过feign调用核心日志服务。底层提供了API&#xff1a; <dependency><groupId>org.qlm</groupId><artifactId>qlm-api</artifactId><version>1.0-SNAPSHOT<…...

如何构建数据驱动的企业?爬虫管理平台是关键桥梁吗?

一、数据驱动时代&#xff1a;为何选择爬虫管理平台&#xff1f; 在信息爆炸的今天&#xff0c;数据驱动已成为企业发展的核心战略之一。爬虫管理平台&#xff0c;作为数据采集的第一站&#xff0c;它的重要性不言而喻。这类平台通过自动化手段&#xff0c;从互联网的各个角落…...

多线程Thread

线程Thread简介 任务、线程、金城、多线程 多任务&#xff1a;短时间切换不同得任务 多线程&#xff1a;通过同一条道路&#xff0c;增加道多条道路&#xff0c;提高使用率&#xff0c;解决堵塞问题 普通方法调多线程只有主线一台执行路径是主线程调run()方法&#xff0c;方…...

计算机网络之WPAN 和 WLAN

上一篇文章内容&#xff1a;无线局域网 1.WPAN&#xff08;无线个人区域网&#xff09; WPAN 是以个人为中心来使用的无线个人区域网&#xff0c;它实际上就是一个低功率、小范围、低速率和低价格的电缆替代技术。 &#xff08;1&#xff09; 蓝牙系统(Bluetooth) &#…...

TikTok海外运营,云手机多种变现方法

从现阶段来看&#xff0c;TikTok 的用户基数不断增长&#xff0c;已然成为全球创业者和品牌的全新竞争舞台。其用户数量近乎 20 亿&#xff0c;年轻用户占据主导&#xff0c;市场渗透率也逐年提高。不管是大型企业、著名品牌&#xff0c;还是个体创业者&#xff0c;都绝不能小觑…...

kubekey在ubuntu24实现kubernetes快速安装

基于Ubunut24.04安装 设置主机名 hostnamectl set-hostname kkmain hostnamectl set-hostname kknode1 hostnamectl set-hostname kknode2关闭swap sudo swapoff -a sudo sed -i s/.*swap.*/#&/ /etc/fstab安装kubekey export KKZONEcn curl -sfL https://get-kk.kubes…...

根据关键词query获取google_img(api方式)

文章目录 说明代码第一部分&#xff1a;链接保存为Json第二部分&#xff1a;链接转换为img 说明 根据关键词query获取google_img USERNAME “xxx” PASSWORD “xxx” 官网申请。 代码 首先获取图片链接&#xff0c;保存为json之后下载。 第一部分&#xff1a;链接保存为…...

西安明德理工学院师生莅临泰迪智能科技开展参观见习活动

为进一步深化校企合作&#xff0c;落实高校应用型人才培养。7月8日&#xff0c;西安明德理工学院与广东泰迪智能科技股份有限公司联合开展学生企业见习活动。西安明德理工学院金融产业学院副院长刘敏、金融学专业负责人张莉萍、金融学专业教师曹艳飞、赵浚妤、泰迪智能科技董事…...

通用机器人里程碑!MIT提出策略组合框架PoCo,解决数据源异构难题,实现机器人多任务灵活执行

18 位人形机器人充当「迎宾」人员&#xff0c;整齐划一向嘉宾挥手&#xff0c;这是 2024 世界人工智能大会上的一个震撼场景&#xff0c;让人们直观感受到了今年机器人的飞速发展。 图源&#xff1a;甲子光年 1954 年&#xff0c;世界上第一台可编程机器人「尤尼梅特」在通用汽…...

基于Java中的SSM框架实现疫情冷链追溯系统项目【项目源码+论文说明】

基于Java中的SSM框架实现疫情冷链追溯系统演示 摘要 近几年随着城镇化发展和居民消费水平的不断提升&#xff0c;人们对健康生活方式的追求意识逐渐加强&#xff0c;生鲜食品逐渐受到大众青睐&#xff0c;诸如盒马鲜生、7-fresh等品牌生鲜超市&#xff0c;一时间如雨后春笋般迅…...

想在vue中预览doxc,excel,pdf文件? vue-office提供包支持

在浩瀚的Vue生态中&#xff0c;vue-office犹如一颗璀璨的星辰&#xff0c;以其独特的魅力照亮了开发者处理多种文件格式的预览之路。这款精心打造的Vue组件库&#xff0c;不仅拥抱了Vue2的经典&#xff0c;也紧密跟随Vue3的步伐&#xff0c;展现了卓越的技术前瞻性和兼容性。它…...

PostgreSQL16安装Mac(brew)

问题 最近需要从MySQL切换到PostgreSQL。我得在本地准备一个PostgreSQL。 步骤 使用brew安装postgresql16: arch -arm64 brew install postgresql16启动postgresql16: brew services start postgresql16配置postgresql环境变量&#xff0c;打开环境变量文件&#xff1a; …...

HTML 语义化

目录 HTML 语义化HTML5 新特性HTML 语义化的好处语义化标签的使用场景最佳实践 HTML 语义化 HTML5 新特性 标准答案&#xff1a; 语义化标签&#xff1a; <header>&#xff1a;页头<nav>&#xff1a;导航<main>&#xff1a;主要内容<article>&#x…...

工业安全零事故的智能守护者:一体化AI智能安防平台

前言&#xff1a; 通过AI视觉技术&#xff0c;为船厂提供全面的安全监控解决方案&#xff0c;涵盖交通违规检测、起重机轨道安全、非法入侵检测、盗窃防范、安全规范执行监控等多个方面&#xff0c;能够实现对应负责人反馈机制&#xff0c;并最终实现数据的统计报表。提升船厂…...

macOS多出来了:Google云端硬盘、YouTube、表格、幻灯片、Gmail、Google文档等应用

文章目录 问题现象问题原因解决办法 问题现象 macOS启动台&#xff08;Launchpad&#xff09;多出来了&#xff1a;Google云端硬盘、YouTube、表格、幻灯片、Gmail、Google文档等应用。 问题原因 很明显&#xff0c;都是Google家的办公全家桶。这些应用并不是通过独立安装的…...

c#开发AI模型对话

AI模型 前面已经介绍了一般AI模型本地部署&#xff0c;直接调用现成的模型数据。这里主要讲述讲接口集成到我们自己的程序中使用方式。 微软提供了ML.NET来开发和使用AI模型&#xff0c;但是目前国内可能使用不多&#xff0c;至少实践例子很少看见。开发训练模型就不介绍了&am…...

在Ubuntu24上采用Wine打开SourceInsight

1. 安装wine sudo apt install wine 2. 安装32位库支持,SourceInsight是32位程序 sudo dpkg --add-architecture i386 sudo apt update sudo apt install wine32:i386 3. 验证安装 wine --version 4. 安装必要的字体和库(解决显示问题) sudo apt install fonts-wqy…...

代码随想录刷题day30

1、零钱兑换II 给你一个整数数组 coins 表示不同面额的硬币&#xff0c;另给一个整数 amount 表示总金额。 请你计算并返回可以凑成总金额的硬币组合数。如果任何硬币组合都无法凑出总金额&#xff0c;返回 0 。 假设每一种面额的硬币有无限个。 题目数据保证结果符合 32 位带…...

Mysql8 忘记密码重置,以及问题解决

1.使用免密登录 找到配置MySQL文件&#xff0c;我的文件路径是/etc/mysql/my.cnf&#xff0c;有的人的是/etc/mysql/mysql.cnf 在里最后加入 skip-grant-tables重启MySQL服务 service mysql restartShutting down MySQL… SUCCESS! Starting MySQL… SUCCESS! 重启成功 2.登…...

node.js的初步学习

那什么是node.js呢&#xff1f; 和JavaScript又是什么关系呢&#xff1f; node.js 提供了 JavaScript的运行环境。当JavaScript作为后端开发语言来说&#xff0c; 需要在node.js的环境上进行当JavaScript作为前端开发语言来说&#xff0c;需要在浏览器的环境上进行 Node.js 可…...

React核心概念:State是什么?如何用useState管理组件自己的数据?

系列回顾&#xff1a; 在上一篇《React入门第一步》中&#xff0c;我们已经成功创建并运行了第一个React项目。我们学会了用Vite初始化项目&#xff0c;并修改了App.jsx组件&#xff0c;让页面显示出我们想要的文字。但是&#xff0c;那个页面是“死”的&#xff0c;它只是静态…...

背包问题双雄:01 背包与完全背包详解(Java 实现)

一、背包问题概述 背包问题是动态规划领域的经典问题&#xff0c;其核心在于如何在有限容量的背包中选择物品&#xff0c;使得总价值最大化。根据物品选择规则的不同&#xff0c;主要分为两类&#xff1a; 01 背包&#xff1a;每件物品最多选 1 次&#xff08;选或不选&#…...