Java版Flink使用指南——将消息写入到RabbitMQ的队列中

大纲

  • 新建工程
    • 新增依赖
  • 编码
    • 自动产生数据
    • 写入RabbitMQ
  • 测试
  • 工程代码

在 《Java版Flink使用指南——从RabbitMQ中队列中接入消息流》一文中,我们介绍了如何使用Java在Flink中读取RabbitMQ中的数据,并将其写入日志中。本文将通过代码产生一些数据,然后将它们写入到另外一个RabbitMQ队列中。

新建工程

我们在IntelliJ中新建一个工程SinkToRabbitMQ。
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>

编码

自动产生数据

这段代码将产生两个字符串数据,后续这些数据会被写入到RabbitMQ的队列中。

List<String> data = new ArrayList<>();
data.add("Hello, World!");
data.add("Hello, Flink!");
DataStream<String> stream = env.fromCollection(data);

写入RabbitMQ

不同于《Java版Flink使用指南——从RabbitMQ中队列中接入消息流》中创建RMQSource用来接收RabbitMQ队列中数据,这次我们创建RMQSink用来发布数据。

String sinkQueueName = "data.to.rbtmq"; // name of the queue to send data to
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;RMQConnectionConfig rmqConnectionConfig = new RMQConnectionConfig.Builder().setHost(host).setPort(port).setUserName(username).setPassword(password).setVirtualHost(virtualHost).build();RMQSink<String> stringRMQSink = new RMQSink<>(rmqConnectionConfig, sinkQueueName, new SimpleStringSchema());
stream.addSink(stringRMQSink).name(username + "'s sink to " + sinkQueueName).setParallelism(parallelism);	

测试

打包、提交并运行任务
在这里插入图片描述
然后在RabbitMQ的后台可以看到收到两条消息
在这里插入图片描述
其内容也是我们之前在代码中生成的内容
在这里插入图片描述

工程代码

https://github.com/f304646673/FlinkDemo

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/pingmian/44924.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

N32G45XVL-STB之lvgl的应用实例

目录 概述 1 硬件介绍 1.1 ST7796-LCD 1.2 MCU IO与LCD PIN对应关系 1.3 MCU IO与Touch PIN对应关系 2 N32G45x移植 LVGL 2.1 移植步骤 2.2 注意点 2.2.1 UI刷新函数 2.2.2 主函数中调用 3 LVGL的应用Demo 3.1 功能描述 3.2 代码实现 3.3 测试 N32G45XVL-STB之lv…

基于RK3588的NPU案例分享!6T是真的强!

RK3588 NPU简介 作为瑞芯微新一代旗舰工业处理器&#xff0c;RK3588 NPU性能可谓十分强大&#xff0c;6TOPS设计能够实现高效的神经网络推理计算。这使得RK3588在图像识别、语音识别、自然语言处理等人工智能领域有着极高的性能表现。 此外&#xff0c;RK3588的NPU还支持多种…

C# 异步编程Invoke、beginInvoke、endInvoke的用法和作用

C# 异步编程Invoke、beginInvoke、endInvoke的用法和作用 一、Invoke Invoke的本质只是一个方法&#xff0c;方法一定是要通过对象来调用的。 一般来说&#xff0c;Invoke其实用法只有两种情况&#xff1a; Control的Invoke Delegate的Invoke 也就是说&#xff0c;Invoke前…

Python基础教学之五:异常处理与文件操作——让程序更健壮

Python基础教学之五&#xff1a;异常处理与文件操作——让程序更健壮 一、异常处理概念 1. 理解异常 异常是程序运行中发生的错误或意外情况&#xff0c;比如除以零、访问不存在的列表元素等。如果不进行处理&#xff0c;异常会导致程序终止运行。在编程过程中&#xff0c;我…

【YOLOv8】 用YOLOv8实现数字式工业仪表智能读数(二)

上一篇圆形表盘指针式仪表的项目受到很多人的关注&#xff0c;咱们一鼓作气&#xff0c;把数字式工业仪表的智能读数也研究一下。本篇主要讲如何用YOLOV8实现数字式工业仪表的自动读数&#xff0c;并将读数结果进行输出&#xff0c;若需要完整数据集和源代码可以私信。 目录 &…

初识Laravel(Laravel的项目搭建)

初识Laravel&#xff08;Laravel的项目搭建&#xff09; 一、项目简单搭建&#xff08;laravel&#xff09;1.首先我们确保使用国内的 Composer 加速镜像&#xff08;[加速原理](https://learnku.com/php/wikis/30594)&#xff09;&#xff1a;2.新建一个名为 Laravel 的项目&a…

简过网:“三支一扶”这些政策你知道吗?

你好小编&#xff0c;我最近打算备考三支一扶&#xff0c;能介绍一些关于三支一扶的相关知识吗&#xff1f; 为了让大家更好的了解三支一扶&#xff0c;下面这篇文章&#xff0c;小编以问答的方式给大家介绍&#xff0c;希望能够帮助到你&#xff01; 1、什么是三支一扶&#…

电脑 DNS 缓存是什么?如何清除?

DNS&#xff08;Domain Name System&#xff0c;域名系统&#xff09;是互联网的重要组成部分&#xff0c;负责将人类易记的域名转换为机器可读的 IP 地址&#xff0c;从而实现网络通信。DNS 缓存是 DNS 系统中的一个关键机制&#xff0c;通过临时存储已解析的域名信息&#xf…

消息队列必知必会-RabbitMQ

文章目录 RabbitMQ是什么&#xff1f;有什么特点&#xff1f;RabbitMQ架构RabbitMQ消息消费过程如何保证消息不丢失&#xff1f;可靠性传输&#xff1f;生产者丢失了数据RabbitMQ&#xff08;broker&#xff09;丢失了数据消费端丢失数据 顺序消息错乱场景解决方案 高可用普通集…

git提交大文件服务500

错误如图 需保证git服务端能接收大文件 修改项目下.git文件中的config文件&#xff0c;加入 [http] postBuffer 524288000

力扣 160相聚链表

注意 判断是否有交点 用while(A! B) 其中A A nullptr? headb:A->next;B同理 注意&#xff0c;while循环的退出条件是AB指针指向同一个&#xff0c;如果没有相交&#xff0c;仍然可以退出 当AB都为NULLPTR时退出

【信创】信创云规划设计建设方案(2024PPT原件)

信创&#xff0c;即“信息技术应用创新”。我国自主信息产业聚焦信息技术应用创新&#xff0c;旨在通过对IT硬件、软件等各个环节的重构&#xff0c;基于我国自有IT底层架构和标准&#xff0c;形成自有开放生态&#xff0c;从根本上解决本质安全问题&#xff0c;实现信息技术可…

【笔试常见编程题06】最近公共祖先、求最大连续bit数、二进制插入、查找组成一个偶数最接近的两个素数

1. 最近公共祖先 将一棵无穷大满二叉树的结点按根结点一层一层地从左往右编号&#xff0c;根结点编号为1。现给定a&#xff0c;b为两个结点。设计一个算法&#xff0c;返回a、b最近的公共祖先的编号。注意其祖先也可能是结点本身。 测试样例&#xff1a; 2&#xff0c;3 返回&a…

Airtest成功案例分享:KLab连续2年携Airtest私有云产品参加CEDEC大会!

一、KLab株式会社介绍 KLab株式会社是一家位于日本的移动游戏开发公司&#xff0c;成立于2000年。公司以开发和运营基于动漫和漫画IP的手机游戏而闻名&#xff0c;尤其是在音乐节奏游戏领域。KLab的一些知名作品包括《LoveLive!学园偶像祭》、《排球少年&#xff1a;新的征程》…

【中项第三版】系统集成项目管理工程师 | 第 11 章 规划过程组② | 11.3 - 11.5

前言 第 11 章对应的内容选择题和案例分析都会进行考查&#xff0c;这一章节属于10大管理的内容&#xff0c;学习要以教材为准。本章上午题分值预计在15分。 目录 11.3 收集需求 11.3.1 主要输入 11.3.2 主要工具与技术 11.3.3 主要输出 11.4 定义范围 11.4.1 主要输入…

【经典面试题】环形链表

1.环形链表oj 2. oj解法 利用快慢指针&#xff1a; /*** Definition for singly-linked list.* struct ListNode {* int val;* struct ListNode *next;* };*/typedef struct ListNode ListNode; bool hasCycle(struct ListNode *head) {ListNode* slow head, *fast…

如何在 C 语言中进行选择排序?

&#x1f345;关注博主&#x1f397;️ 带你畅游技术世界&#xff0c;不错过每一次成长机会&#xff01; &#x1f4d9;C 语言百万年薪修炼课程 通俗易懂&#xff0c;深入浅出&#xff0c;匠心打磨&#xff0c;死磕细节&#xff0c;6年迭代&#xff0c;看过的人都说好。 文章目…

【Python】已解决:AttributeError: target_names(机器学习中查看数据信息报错)

文章目录 一、分析问题背景二、可能出错的原因三、错误代码示例四、正确代码示例五、注意事项 已解决&#xff1a;AttributeError: target_names&#xff08;机器学习中查看数据信息报错&#xff09; 一、分析问题背景 在机器学习的数据处理阶段&#xff0c;我们经常需要查看…

使用pip或conda离线下载安装包,使用pip或conda安装离线安装包

使用pip或conda离线下载安装包&#xff0c;使用pip或conda安装离线安装包 一、使用pip离线下载安装包1. 在有网络的机器上下载包和依赖2. 传输离线安装包 二、在目标机器上离线安装pip包三、使用conda离线下载安装包1. 在有网络的机器上下载conda包2. 传输conda包或环境包3. 在…

每日一练 - 理解IGMP组播组信息

下面是路由器 RTB 的部分输出信息&#xff0c; 关于输出信息描述错误的是A.接口上动态加入的组播组个数是 1 B.加入的组播组地址是 225.1.1.2 C.dsplay igmp group 命令用来查看 IGMP 组播组信息,包括通过成员报告动态加入的组播组和通过命令行静态加入的组播组信息 D.最后发…