0

我想使用 Siddhi 流处理器从 NATS 服务器读取消息。我已经安装了以下软件:

  • NATS 服务器 v2.1.6(在独立服务器上)
  • Siddhi siddhi-runner-5.1.2(作为同一 LAN 上另一台服务器上的独立二进制文件)
  • NATS 2.0.10 的 Siddhi 扩展,具有以下依赖项:protobuf-java-3.9.1.jar、java-nats-streaming-2.2.2.jar、jnats-2.6.5.jar(反正我不使用流选项)

我使用以下 siddhi 应用程序:

@App:name("TestNATS")

@source(type='nats', @map(type='text'), destination='test.message', bootstrap.servers='nats://192.168.50.173:4222')
define stream inputStream (name string, age int, country string);

@sink(type='log', prefix='LOGGER')
define stream OutputStream(name string, age int, country string);

@info(name='HelloWorldQueryNATS')
from inputStream
select name, age, country
insert into OutputStream;

当 Siddhi 启动时,它似乎可以NATS server毫无问题地连接,但在最初的握手之后,Siddhi 会抛出以下内容exception

[2020-04-10 10:55:00,441]  INFO {org.wso2.msf4j.internal.websocket.WebSocketServerSC} - All required capabilities are available of WebSocket service component is available.

[2020-04-10 10:55:00,549]  INFO {org.wso2.msf4j.analytics.metrics.MetricsComponent} - Metrics Component is activated

[2020-04-10 10:55:00,552]  INFO {org.wso2.carbon.databridge.agent.internal.DataAgentDS} - Successfully deployed Agent Server

[2020-04-10 10:55:01,163]  INFO {org.wso2.carbon.analytics.idp.client.core.utils.IdPServiceUtils} - Enabling default IdPClient Local User Store as configuration is not overridden.

[2020-04-10 10:55:01,167]  INFO {org.wso2.carbon.analytics.idp.client.core.utils.IdPServiceUtils} - IdP client of type 'local' is started.

[2020-04-10 10:55:06,915] ERROR {io.siddhi.extension.io.nats.source.NATSSource} - Error while connecting to NATS server at destination: test.message

[2020-04-10 10:55:06,918] ERROR {io.siddhi.core.stream.input.source.Source} - Error on 'TestNATS'. Error while connecting to NATS server at destination: test.message Error while connecting at Source 'nats' at 'inputStream'. Will retry in '5 sec'. io.siddhi.core.exception.ConnectionUnavailableException: Error while connecting to NATS server at destination: test.message


Caused by: java.io.IOException: stan: connect request timeout

并不断重启连接过程,所以在日志中NATS server我们可以看到如下信息:

cid:1 - Client connection created

cid:1 - <<- [CONNECT"lang":"java","version":"2.6.0","name":"1586508901596_66035_83044","protocol":1,"verbose":false,"pedantic":false,"tls_required":false,"echo":true}]

cid:1 - <<- [PING]

cid:1 - ->> [PONG]

cid:1 - <<- [SUB _STAN.acks.TSHykWgo2B6Lv2xW8b1Bfd 1]

cid:1 - <<- [SUB _INBOX.TSHykWgo2B6Lv2xW8b1BaH 2]

cid:1 - <<- [SUB _INBOX.TSHykWgo2B6Lv2xW8b1Bcx 3]

cid:1 - <<- [SUB _INBOX.4bm5CWKphBwldfvnGCxxJd.* 4]

cid:1 - <<- [PUB _STAN.discover.test-cluster _INBOX.4bm5CWKphBwldfvnGCxxJd.4bm5CWKphBwldfvnGCxxa9 88]

cid:1 - <<- MSG_PAYLOAD: ["\n\x191586508901596_66035_83044\x12\x1d_INBOX.TSHykWgo2B6Lv2xW8b1BaH\x18\x01\"\x16TSHykWgo2B6Lv2xW8b1BXb(\x050\x03"]

cid:1 - Client Ping Timer

cid:1 - ->> [PING]

cid:1 - <<- [PONG]

cid:1 - <<- [UNSUB 1]

cid:1 - <-> [DELSUB 1]

cid:1 - <<- [UNSUB 2]

cid:1 - <-> [DELSUB 2]

cid:1 - <<- [UNSUB 3]

cid:1 - <-> [DELSUB 3]

cid:1 - Client connection closed

cid:1 - <-> [DELSUB 4]

cid:2 - Client connection created

NATS server与我在 pub-sub 应用程序中使用的相同Nodejs,它工作正常。我也尝试使用最后一个 Siddhidocker包,但结果是一样的。

哪里可能出错?

4

1 回答 1

0

使用 nats-streaming-server 代替 nats-server,不再出现连接问题。谢谢。

于 2020-04-11T14:46:51.063 回答