RTopic’s subscription function always fails randomly about redisson HOT 9 OPEN

Unable to reproduce with the test below. What did I miss?

        int loops = 10;
        AtomicInteger counter = new AtomicInteger();
        for (int j = 0; j < loops; j++) {
            RTopic t = redisson.getTopic("PUBSUB_" + j);
            t.addListener(String.class, new MessageListener<String>() {
                @Override
                public void onMessage(CharSequence channel, String msg) {
                    System.out.println("channel " + channel + " " + msg);
                    counter.incrementAndGet();
                }
            });
        }

        for (int j = 0; j < loops; j++) {
            RTopic t = redisson.getTopic("PUBSUB_" + j);
            for (int i = 0; i < 10; i++) {
                t.publish("message " + j + "_" + i);
            }
        }

        Thread.sleep(1000);

        assertThat(counter.get()).isEqualTo(loops * 10);

from redisson.

It is not easy to reproduce in a short period of time. You need to continue publishing messages, and then after running for a period of time, you will find that several channels cannot be consumed.

Unable to reproduce with the test below. What did I miss?

from redisson.

I don't know the specific reason. In my scenario: the same redis, the same number of channels, the same publish, the same configuration, and different Redisson versions.

I once suspected that the redis buffer overflow caused the client connection to be disconnected. I used the client list command to check the buffer usage and found that the buffer was very sufficient, thus ruling out the possibility of the redis server actively disconnecting.

Then I tried to add the Status listener of the Topic, and found that when the count bit of the publish result was 0, the status listener did not trigger any unsubscribe event. I tried repeatedly by switching different Redisson versions, and found that only 3.25.1 had no problem, so I had to Roll back to previous version.

from redisson.

Can you try 3.26.0?

from redisson.

Can you try 3.26.0?

I can test it later when I have time, but I'm sorry not now.

from redisson.

do you invoke RTopic.addListener() only once per channel? Total 10 channels?

from redisson.

This is my code. Each channel only triggers this code once.

        RTopic rt = getTopic(topic);

        rt.addListener(new StatusListener() {
            @Override
            public void onSubscribe(String channel) {
                logger.debug("redis channel subscribed {}", channel);
            }

            @Override
            public void onUnsubscribe(String channel) {
                logger.info("redis channel unsubscribed {}", channel);
            }
        });

        rt.addListener(
                TransformMessage.class,
                (channel, msg) -> executor.execute(() -> consume(MQMessage.create(channel.toString(), msg), consumer))
        );

from redisson.

Do you use a single Redis server or cluster setup?

from redisson.

No, my production environment uses sentinel mode.

Do you use a single Redis server or cluster setup?

from redisson.