brainlag / javansqclient Goto Github PK
View Code? Open in Web Editor NEWThis project forked from nsqio/trendrrnsqclient
Fast Java client for NSQ
License: MIT License
This project forked from nsqio/trendrrnsqclient
Fast Java client for NSQ
License: MIT License
Can I ask about when version 1.0.0 will be released ? I am planning to use this project with the one of the project I am contributing and I'm uneasy at the idea of having an RC as dependency.
While doing a connection close, there is an Channel Inactive which is un-caught. Hence, the connection is not removed.
So I'm left with a stale connection, and not able to reconnect due to an existing entry to that address.
Logs:
16:14:47.245 INFO Remove connection server:4150 - NSQConsumer.connect
16:14:47.245 INFO Closing connection: com.github.brainlag.nsq.Connection@188fe003 - Connection.close
16:14:47.246 INFO Channel disconnected! com.github.brainlag.nsq.Connection@188fe003 - NSQHandler.channelInactive
Fix could be to put in try catch
for (final ServerAddress server : Sets.difference(oldAddresses, newAddresses)) {
LogManager.getLogger(this).info("Remove connection " + server.toString());
try {
connections.get(server).close();
} catch (Exception e) {
LogManager.getLogger(this).error("Close connection failed: " + e.getMessage(), e);
} finally {
connections.remove(server);
}
}
Producer works fine, but consumer can not start . Stacktrace may like this:
com.github.brainlag.nsq.exceptions.NoConnectionsException: Could not connect to server
at com.github.brainlag.nsq.Connection.(Connection.java:62) ~[nsq-client-1.0.0.RC4.jar:?]
at com.github.brainlag.nsq.NSQConsumer.createConnection(NSQConsumer.java:80) ~[nsq-client-1.0.0.RC4.jar:?]
at com.github.brainlag.nsq.NSQConsumer.connect(NSQConsumer.java:210) ~[nsq-client-1.0.0.RC4.jar:?]
at com.github.brainlag.nsq.NSQConsumer.lambda$start$0(NSQConsumer.java:73) ~[nsq-client-1.0.0.RC4.jar:?]
at java.util.concurrent.Executors$RunnableAdapter.call$$$capture(Executors.java:511) [?:1.8.0_144]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java) [?:1.8.0_144]
at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308) [?:1.8.0_144]
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180) [?:1.8.0_144]
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294) [?:1.8.0_144]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [?:1.8.0_144]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [?:1.8.0_144]
at java.lang.Thread.run(Thread.java:748) [?:1.8.0_144]
Caused by: java.nio.channels.UnresolvedAddressException
at sun.nio.ch.Net.checkAddress(Net.java:101) ~[?:1.8.0_144]
at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:622) ~[?:1.8.0_144]
at io.netty.channel.socket.nio.NioSocketChannel.doConnect(NioSocketChannel.java:208) ~[netty-all-4.0.39.Final.jar:4.0.39.Final]
at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.connect(AbstractNioChannel.java:203) ~[netty-all-4.0.39.Final.jar:4.0.39.Final]
at io.netty.channel.DefaultChannelPipeline$HeadContext.connect(DefaultChannelPipeline.java:1226) ~[netty-all-4.0.39.Final.jar:4.0.39.Final]
at io.netty.channel.AbstractChannelHandlerContext.invokeConnect(AbstractChannelHandlerContext.java:549) ~[netty-all-4.0.39.Final.jar:4.0.39.Final]
at io.netty.channel.AbstractChannelHandlerContext.connect(AbstractChannelHandlerContext.java:534) ~[netty-all-4.0.39.Final.jar:4.0.39.Final]
at io.netty.channel.ChannelOutboundHandlerAdapter.connect(ChannelOutboundHandlerAdapter.java:47) ~[netty-all-4.0.39.Final.jar:4.0.39.Final]
at io.netty.channel.AbstractChannelHandlerContext.invokeConnect(AbstractChannelHandlerContext.java:549) ~[netty-all-4.0.39.Final.jar:4.0.39.Final]
at io.netty.channel.AbstractChannelHandlerContext.connect(AbstractChannelHandlerContext.java:534) ~[netty-all-4.0.39.Final.jar:4.0.39.Final]
at io.netty.channel.AbstractChannelHandlerContext.connect(AbstractChannelHandlerContext.java:516) ~[netty-all-4.0.39.Final.jar:4.0.39.Final]
at io.netty.channel.DefaultChannelPipeline.connect(DefaultChannelPipeline.java:970) ~[netty-all-4.0.39.Final.jar:4.0.39.Final]
at io.netty.channel.AbstractChannel.connect(AbstractChannel.java:215) ~[netty-all-4.0.39.Final.jar:4.0.39.Final]
at io.netty.bootstrap.Bootstrap$2.run(Bootstrap.java:166) ~[netty-all-4.0.39.Final.jar:4.0.39.Final]
at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:408) ~[netty-all-4.0.39.Final.jar:4.0.39.Final]
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:402) ~[netty-all-4.0.39.Final.jar:4.0.39.Final]
at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:140) ~[netty-all-4.0.39.Final.jar:4.0.39.Final]
at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:144) ~[netty-all-4.0.39.Final.jar:4.0.39.Final]
... 1 more
How can I fix this?
There are breaking changes between 4.1 and 4.0 Most of the project start to use 4.1 and this creates classpath errors. It'll be good upgrading netty to 4.1.x
The following error happens simply when reading messages for a while from NSQConsumer.
[nioEventLoopGroup-2-2] ERROR io.netty.util.ResourceLeakDetector - LEAK: ByteBuf.release() was not called before it's garbage-collected. Enable advanced leak reporting to find out where the leak occurred. To enable advanced leak reporting, specify the JVM option '-Dio.netty.leakDetection.level=advanced' or call ResourceLeakDetector.setLevel() See http://netty.io/wiki/reference-counted-objects.html for more information.
Consumer never be triggered about network error. For example, unplug the network cable, disable computer networks. so connections in consumer never be closed and remove.
I think should add heartbeat interval expire for connection. If last heartbeat interval exceeded, close and reconnection.
I will post my code improvements soon.
Hi!brainlag.
I use your 'nsq-client' in my clojure-project. But I can not connect to lookup. The code is as follow:
(import com.github.brainlag.nsq.lookup.DefaultNSQLookup)
=> com.github.brainlag.nsq.lookup.DefaultNSQLookup
(def nsq-lookup (DefaultNSQLookup.))
=> #'nsq-tester.core/nsq-lookup
(.addLookupAddress nsq-lookup "127.0.0.1" 4161)
=> nil
(.lookup nsq-lookup "testss")
17:28:42.004 WARN Unable to connect to address http://127.0.0.1:4161 - DefaultNSQLookup.lookup
17:28:42.004 WARN Unable to connect to any NSQ Lookup servers, servers tried: [http://127.0.0.1:4161] - DefaultNSQLookup.lookup
=> #{}
I run this code on the repl.
the nsq is deployed on my PC.
i run testing testProduceOneMsgSnappy,
producer works, but consumer doesn't work, the callback won't get any message.
Hi, nice to find this nsq java-client, I want to use it for production, so I have that doubt..
Connection.java, lines 70, 71
final NSQFrame response = commandAndWait(ident);
LogManager.getLogger(this).info("Server identification: " + ((ResponseFrame) response).getMessage());
commandAndWait returns null on an InterruptedException, which causes the log line to throw an NPE on response.getMessage()
I haven't been able to reproduce this error yet, but I saw it while testing.
Any plans to implement auth in this library?
I'd like to sf4j-log4j to instead of log4j2, cause sf4j is a simple Logging Facade and could be used in various logging frameworks (e.g. java.util.logging, logback, log4j) .
If you think that's needed, I will open a merge request to you.
nsq version:1.0
os:centos 7.4
codes:
public static void main( String[] args )
{
NSQLookup lookup = new DefaultNSQLookup();
lookup.addLookupAddress("192.168.1.228", 4161);
NSQConsumer consumer = new NSQConsumer(lookup, "TestTopic", "dusti", (message) -> {
System.out.println("received: " + new String(message.getMessage()));
message.finished();
message.requeue();
});
consumer.start();
}
below is run time messages:
17:00:37.080 INFO Created connection: knowledgebase:4150 - Connection.
17:00:37.100 INFO IdentifyResponse: {"max_rdy_count":2500,"version":"1.0.0-compat","max_msg_timeout":900000,"msg_timeout":60000,"tls_v1":false,"deflate":false,"deflate_level":0,"max_deflate_level":6,"snappy":false,"sample_rate":0,"auth_required":false,"output_buffer_size":16384,"output_buffer_timeout":250} - NSQFeatureDetectionHandler.channelRead0
17:00:37.101 INFO reinstall LengthFieldBasedFrameDecoder - NSQFeatureDetectionHandler.eject
17:00:37.102 INFO Server identification: {"max_rdy_count":2500,"version":"1.0.0-compat","max_msg_timeout":900000,"msg_timeout":60000,"tls_v1":false,"deflate":false,"deflate_level":0,"max_deflate_level":6,"snappy":false,"sample_rate":0,"auth_required":false,"output_buffer_size":16384,"output_buffer_timeout":250} - Connection.
received: test8
received: test9
十二月 28, 2017 5:00:53 下午 io.netty.util.concurrent.SingleThreadEventExecutor runAllTasks
警告: A task raised an exception.
java.lang.IllegalStateException: Queue full
at java.util.AbstractQueue.add(AbstractQueue.java:98)
at com.github.brainlag.nsq.Connection.incoming(Connection.java:129)
at com.github.brainlag.nsq.netty.NSQHandler.lambda$channelRead0$3(NSQHandler.java:41)
at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:408)
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:402)
at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:140)
at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:144)
at java.lang.Thread.run(Thread.java:748)
Currently all connections use the same EventLoopGroup which might be too limiting for some use cases. Therefore we should expose an API to allow the user to control the EventLoopGroup for each connection.
I try to set the timeout of NSQConsumer but this throw a Cast Exception
NSQConfig config= new NSQConfig();
config.setMsgTimeout(Util.NSQ_MSG_TIMEOUT);
NSQConsumer consumer = new NSQConsumer(lookup, topic, NsqChannelConst.REPORTS_CHANNEL, (message) -> {
//......
},config);
this is the error:
Caused by: java.lang.ClassCastException: class com.github.brainlag.nsq.frames.ErrorFrame cannot be cast to class com.github.brainlag.nsq.frames.ResponseFrame (com.github.brainlag.nsq.frames.ErrorFrame and com.github.brainlag.nsq.frames.ResponseFrame are in unnamed module of loader 'app')
at com.github.brainlag.nsq.Connection.(Connection.java:77) ~[nsq-client-1.0.0.RC4.jar:na]
at com.github.brainlag.nsq.NSQConsumer.createConnection(NSQConsumer.java:80) ~[nsq-client-1.0.0.RC4.jar:na]
at com.github.brainlag.nsq.NSQConsumer.connect(NSQConsumer.java:210) ~[nsq-client-1.0.0.RC4.jar:na]
at com.github.brainlag.nsq.NSQConsumer.start(NSQConsumer.java:72) ~[nsq-client-1.0.0.RC4.jar:na]
I found that after I call the shutdown() method of the NSQConsumer, there is still one thread of the netty nioEventLoopGroup
It looks like this code from NSQConsumer.java (around line 202) will sometimes throw a ConcurrentModificationException:
for (final ServerAddress server : Sets.difference(oldAddresses, newAddresses)) {
LogManager.getLogger(this).info("Remove connection " + server.toString());
connections.get(server).close();
connections.remove(server);
}
Since the exception is uncaught, the scheduler.scheduleAtFixedRate then silently fails, which is bad.
I use android studio run this test .
java.lang.NoClassDefFoundError: Failed resolution of: Ljava/util/Optional;
//curl -d 'message 1' 'http://localhost:4151/put?topic=test_snappy'
//curl -d 'message 2' 'http://localhost:4151/put?topic=test_snappy'
//curl -d 'message 3' 'http://localhost:4151/put?topic=test_snappy'
NSQLookup lookup = new DefaultNSQLookup();
lookup.addLookupAddress("localhost", 4161);
NSQConfig config = new NSQConfig();
config.setCompression(NSQConfig.Compression.SNAPPY);
NSQConsumer consumer = new NSQConsumer(lookup, "test_snappy", "test_consume",
(msg) -> {
System.out.println(new String(msg.getMessage()));
msg.finished();
}, config);
consumer.start();
Thread.sleep(20000);
System.out.println("done");
10:10:01.997 INFO Created connection: Rob-Seeds-MacBook-Pro.local:4150 - Connection.<init>
10:10:02.021 INFO IdentifyResponse: {"max_rdy_count":2500,"version":"0.3.2","max_msg_timeout":900000,"msg_timeout":60000,"tls_v1":false,"deflate":false,"deflate_level":0,"max_deflate_level":6,"snappy":true,"sample_rate":0,"auth_required":false,"output_buffer_size":16384,"output_buffer_timeout":250} - NSQFeatureDetectionHandler.channelRead0
10:10:02.025 INFO Adding snappy to pipline - NSQFeatureDetectionHandler.installSnappyDecoder
10:10:02.027 INFO IdentifyResponse: OK - NSQFeatureDetectionHandler.channelRead0
10:10:02.029 INFO Server identification: {"max_rdy_count":2500,"version":"0.3.2","max_msg_timeout":900000,"msg_timeout":60000,"tls_v1":false,"deflate":false,"deflate_level":0,"max_deflate_level":6,"snappy":true,"sample_rate":0,"auth_required":false,"output_buffer_size":16384,"output_buffer_timeout":250} - Connection.<init>
message 1
done
Works as expected when not using snappy compression. Publishing with snappy enabled works fine.
I've created a consumer and a producer, and had to jump through some hoops to get the process to close. I ended up having to:
It would be nice if this were either easier to do, or documented.
JsonNode jsonNode = mapper.readTree(new URL(addr + "/lookup?topic=" + topicEncoded)); LogManager.getLogger(this).debug("Server connection information: {}", jsonNode); JsonNode producers = jsonNode.get("data").get("producers");
jsonNode value is not having "data" key but rather "producers" is in the same level:
{"channels":["test1"],"producers":[{"remote_address":"127.0.0.1:52354","hostname":"vikash.tiwari","broadcast_address":"vikash.tiwari","tcp_port":4150,"http_port":4151,"version":"1.0.0-compat"}]}
Producer and Consumer connection code is same as in instructions.
Hi, I am wondering how to connect to nsqd without lookup. The consumer samples online always use nsqdlookup firstly then connect and I've failed to find a way to set connection in NSQConsumer. Thank you.
Because the use of jdk1.7 so I want to use RPC framework(JDK1.7), switching effects will be great? If it is how to do it?
java.lang.IllegalStateException: Queue full
at java.util.AbstractQueue.add(AbstractQueue.java:98)
at com.github.brainlag.nsq.Connection.incoming(Connection.java:115)
at com.github.brainlag.nsq.netty.NSQHandler.lambda$channelRead0$3(NSQHandler.java:41)
at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:358)
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:357)
at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:112)
at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:137)
at java.lang.Thread.run(Thread.java:748)
Went through the code and found the following line in Connection.java :
private final LinkedBlockingQueue<NSQFrame> responses = new LinkedBlockingQueue<>(1);
Wanted to understand why this is initialized with capacity 1 and what could potentially be causing this error. Could it be related to high rate of polling NSQ? We deal with a lot of data.
Hi,
As the title said, I wanna use NSQ to implement a simple service bus, it is very easy ti find that the pub/sub and queue way have already implemented in NSQ, but how to get return object in a NSQProducer? Or do you have any kindly subjection?
Best regards.
Why the implementation of producer not apply consumer model, get nsqd server address by lookup lookupd servers, then establish a connection for message sending, and decide which nsqd to send a message by scheduling lookup lookupd servers
like this:
`
public NSQConsumer(final NSQLookup lookup, final String topic, final String channel, final NSQMessageCallback callback) {
this(lookup, topic, channel, callback, new NSQConfig());
}
public NSQConsumer(final NSQLookup lookup, final String topic, final String channel, final NSQMessageCallback callback,
final NSQConfig config) {
this(lookup, topic, channel, callback, config, null);
}
public NSQConsumer(final NSQLookup lookup, final String topic, final String channel, final NSQMessageCallback callback,
final NSQConfig config, final NSQErrorCallback errCallback) {
this.lookup = lookup;
this.topic = topic;
this.channel = channel;
this.config = config;
this.callback = callback;
this.errorCallback = errCallback;
}
`
but Producer has initialized hardly like this:
`
public class NSQProducer {
private Set addresses = Sets.newConcurrentHashSet();
...
protected Connection getConnection() throws NoConnectionsException {
int c = 0;
while (c < connectionRetries) {
ServerAddress[] serverAddresses = addresses.toArray(new ServerAddress[addresses.size()]);
if (serverAddresses.length != 0) {
try {
return pool.borrowObject(serverAddresses[roundRobinCount++ % serverAddresses.length]);
} catch (NoSuchElementException e) {
try {
Thread.sleep(1000);
} catch (InterruptedException ix) {
throw new NoConnectionsException("Could not acquire a connection to a server", ix);
}
} catch (Exception ex) {
throw new NoConnectionsException("Could not acquire a connection to a server", ex);
}
}
}
throw new IllegalStateException("No server configured for producer");
}
`
I think it's good for Increase/Decrease number of nsqd servers.
A declarative, efficient, and flexible JavaScript library for building user interfaces.
🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
The Web framework for perfectionists with deadlines.
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. 📊📈🎉
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
We are working to build community through open source technology. NB: members must have two-factor auth.
Open source projects and samples from Microsoft.
Google ❤️ Open Source for everyone.
Alibaba Open Source for everyone
Data-Driven Documents codes.
China tencent open source team.