START TXN 123 --No message sent to Kafka
123 - Event1 Message --No message sent to Kafka
123 - Event2 Message --No message sent to Kafka
123 - Event3 Message --No message sent to Kafka
COMMIT TXN 123 --Event1, Event2, Event3 messages sent to Kafka
input {
tcp {
port => 9000
}
}
output {
kafka {
bootstrap_servers => "localhost:9092"
topic_id => "alpayk"
}
}
我试着使用use logstash的聚合过滤器来达到这个目的,但是我不能得到一个工作的结果。
非常感谢你
我最终决定为此使用Apache水槽。我修改了它的netcat源,使未提交的消息驻留在Flume的堆中,并且一旦收到事务的提交消息,所有消息就会发送到kafka接收器。
我将把消息存储位置从flume堆更改为外部缓存,这样,如果事务停止或回滚,我将能够使存储的消息过期。
下面是该事务逻辑的代码:
String eventMessage = new String(body);
int indexOfTrxIdSeparator = eventMessage.indexOf("-");
if (indexOfTrxIdSeparator != -1) {
String txnId = eventMessage.substring(0, indexOfTrxIdSeparator).trim();
String message = eventMessage.substring(indexOfTrxIdSeparator + 1).trim();
ArrayList<Event> events = cachedEvents.get(txnId);
if (message.equals("COMMIT")) {
System.out.println("@@@@@ COMMIT RECEIVED");
if (events != null) {
for (Event eventItem : events) {
ChannelException ex = null;
try {
source.getChannelProcessor().processEvent(eventItem);
} catch (ChannelException chEx) {
ex = chEx;
}
if (ex == null) {
counterGroup.incrementAndGet("events.processed");
} else {
counterGroup.incrementAndGet("events.failed");
logger.warn("Error processing event. Exception follows.", ex);
}
}
cachedEvents.remove(txnId);
}
} else {
System.out.println("@@@@@ MESSAGE RECEIVED: " + message);
if (events == null) {
events = new ArrayList<Event>();
}
events.add(EventBuilder.withBody(message.getBytes()));
cachedEvents.put(txnId, events);
}
}
我有一个后端服务器,它将事件作为服务器发送的事件发送给客户端。我还没有找到一个好的库来在Android上处理这项技术,所以我一直在使用一种回退方法,定期检查服务器(通过访问事件endpoint)中的新事件。 后台服务每10秒执行一次。不用说,这不是最好的方法。如果没有任何开源库可用于此场景,那么在内存使用和电池消耗方面,定期检查服务器后端是否有新事件的最佳方法是什么?与在Android中管理开放式
概述 客户端代码 概述 建立连接 open事件 message事件 error事件 自定义事件 close方法 数据格式 概述 data:数据栏 id:数据标识符 event栏:自定义信息类型 retry:最大间隔时间 服务器代码 参考链接 概述 传统的网页都是浏览器向服务器“查询”数据,但是很多场合,最有效的方式是服务器向浏览器“发送”数据。比如,每当收到新的电子邮件,服务器就向浏览器发送一个“
概述 客户端代码 概述 建立连接 open事件 message事件 error事件 自定义事件 close方法 数据格式 概述 data:数据栏 id:数据标识符 event栏:自定义信息类型 retry:最大间隔时间 服务器代码 参考链接 概述 传统的网页都是浏览器向服务器“查询”数据,但是很多场合,最有效的方式是服务器向浏览器“发送”数据。比如,每当收到新的电子邮件,服务器就向浏览器发送一个“
我的登录格式如下,它是一个带有嵌套字段的普通json。 如何使用Filebeat和Logstash正确地解析它,以将Kibana中的所有json字段视为单独的(已解析的)字段?我在“message”字段中遇到了一个问题,它嵌套了json字段。我解析一个在“message”中有字符串的事件没有问题,但不是JSON。 我的尝试: [2019-03-08T09:55:47,084][WARN][logs
当前,我已经在Express API上为服务器发送的事件设置了一个endpoint,当我从cmd调用endpoint时,使用http://localhost:3000/v1/devices,我得到了流,但当我使用新事件源从Vue应用程序调用endpoint时('http://localhost:3000/v1/devices“)但它只会继续加载,而不会流式传输任何数据。是否有我目前没有做的事情要做
我试图让服务器发送的事件与Mozilla Firefox一起工作。给定一个Spring Boot的网络服务 使用Chrome浏览器或Edge(始终是最新版本)可以正常工作。我可以在网络分析器选项卡中看到未完成的请求,并且每秒都会显示一个新的时间戳。 然而,当我使用Firefox(84.0.2或更早版本)时,请求也会显示在网络选项卡中,但不会显示响应头或流数据。当我终止Spring后端时,Firef