Spring: Single threaded consumer

Some workers (listeners) have to handle messages one by one in single-thread-style. (Yea, this is weak point, I know.) And looks, I dont know how to set it up right. Application hangs on Spring5 and this block me to move from Spring4.

I want to highlight: application should have multiple single threaded listeners. One listener for one queue. I mean some number of . Not multiple listeners for one queue.

I use kotlin. But the trouble is in configuration or even in library bug.

@RabbitListener(

        queues = [(queueName1)],

        containerFactory = "exclusiveListenerContainerFactory"

    )

fun handle1(...



@RabbitListener(

        queues = [(queueName2)],

        containerFactory = "exclusiveListenerContainerFactory"

    )

fun handle2(...



@Bean

open fun exclusiveListenerContainerFactory(

    connectionFactory: ConnectionFactory,

    messageConverter: MessageConverter

): SimpleRabbitListenerContainerFactory {

    val factory = SimpleRabbitListenerContainerFactory()

    factory.setConnectionFactory(connectionFactory)

    factory.setConcurrentConsumers(1)

    factory.setMessageConverter(messageConverter)

    factory.setDefaultRequeueRejected(false)

    return factory

}

I used SINGLE exclusiveListenerContainerFactory for each kind on listeners.

And SINGLE rabbitConnectionFactory (instance of CachingConnectionFactory)

I dont know is this right.

Trouble is:

Some times application hangs on Spring5. On Spring 4 same configuration work well. Funny, that it hands on machine with KVM virtualization. With Virtualbox work well even with Spring5.

jstack https://pastebin.com/MXdHcVAK

top -H -p shows PID 42 consumes all cpu

Some time ago I tried to move to Spring5 with same trouble. I thought it was reactor-netty lib issue, but no luck, they can't reproduce https://github.com/reactor/reactor-netty/issues/381

I think these two things - configuration and hanging - is bound. But I can't get deeper.

Update

About dump:
If I understand everything right (I am not sure), cpu consumed by "tcp-client-loop-nio-4" #24 daemon prio=5 os_prio=0 tid=0x00007fc0de44d800 nid=0x2a runnable entry in jstack. But I don't know what to do with that.

Update2

Lead developer of Spring Integration and Spring AMQP - @GaryRussell - approve this configuration. So this issue is about netty I think.

edited Dec 18 '18 at 20:00

Brian Clozel

32k779104

asked Nov 22 '18 at 12:20

Lewik

597

Take a thread dump when it hangs to see what the container threads are doing. If you can't figure it out add the dump to the question.

– Gary Russell
Nov 22 '18 at 15:32

@GaryRussell it's already here jstack pastebin.com/MXdHcVAK Is my configuration correct?

– Lewik
Nov 22 '18 at 15:43

Why do you think this is related to RabbitMQ? The container threads are all either blocked waiting for a new message or blocked in reactor code via Stomp. Looks unrelated to RabbitMQ to me.

– Gary Russell
Nov 22 '18 at 15:53

I definitely don't know source of lock. I assumed this could be rabbitmq. I thought netty used for rabbit.

– Lewik
Nov 22 '18 at 15:57

1

Yes it's fine; but concurrentConsumers defaults to 1 anyway.

– Gary Russell
Nov 22 '18 at 16:27

|
show 4 more comments

I want to highlight: application should have multiple single threaded listeners. One listener for one queue. I mean some number of . Not multiple listeners for one queue.

I use kotlin. But the trouble is in configuration or even in library bug.

@RabbitListener(

        queues = [(queueName1)],

        containerFactory = "exclusiveListenerContainerFactory"

    )

fun handle1(...



@RabbitListener(

        queues = [(queueName2)],

        containerFactory = "exclusiveListenerContainerFactory"

    )

fun handle2(...



@Bean

open fun exclusiveListenerContainerFactory(

    connectionFactory: ConnectionFactory,

    messageConverter: MessageConverter

): SimpleRabbitListenerContainerFactory {

    val factory = SimpleRabbitListenerContainerFactory()

    factory.setConnectionFactory(connectionFactory)

    factory.setConcurrentConsumers(1)

    factory.setMessageConverter(messageConverter)

    factory.setDefaultRequeueRejected(false)

    return factory

}

I used SINGLE exclusiveListenerContainerFactory for each kind on listeners.

And SINGLE rabbitConnectionFactory (instance of CachingConnectionFactory)

I dont know is this right.

Trouble is:

Some times application hangs on Spring5. On Spring 4 same configuration work well. Funny, that it hands on machine with KVM virtualization. With Virtualbox work well even with Spring5.

jstack https://pastebin.com/MXdHcVAK

top -H -p shows PID 42 consumes all cpu

Some time ago I tried to move to Spring5 with same trouble. I thought it was reactor-netty lib issue, but no luck, they can't reproduce https://github.com/reactor/reactor-netty/issues/381

I think these two things - configuration and hanging - is bound. But I can't get deeper.

Update

Update2

Lead developer of Spring Integration and Spring AMQP - @GaryRussell - approve this configuration. So this issue is about netty I think.

edited Dec 18 '18 at 20:00

Brian Clozel

32k779104

asked Nov 22 '18 at 12:20

Lewik

597

Take a thread dump when it hangs to see what the container threads are doing. If you can't figure it out add the dump to the question.

– Gary Russell
Nov 22 '18 at 15:32

@GaryRussell it's already here jstack pastebin.com/MXdHcVAK Is my configuration correct?

– Lewik
Nov 22 '18 at 15:43

Why do you think this is related to RabbitMQ? The container threads are all either blocked waiting for a new message or blocked in reactor code via Stomp. Looks unrelated to RabbitMQ to me.

– Gary Russell
Nov 22 '18 at 15:53

I definitely don't know source of lock. I assumed this could be rabbitmq. I thought netty used for rabbit.

– Lewik
Nov 22 '18 at 15:57

1

Yes it's fine; but concurrentConsumers defaults to 1 anyway.

– Gary Russell
Nov 22 '18 at 16:27

|
show 4 more comments

I want to highlight: application should have multiple single threaded listeners. One listener for one queue. I mean some number of . Not multiple listeners for one queue.

I use kotlin. But the trouble is in configuration or even in library bug.

@RabbitListener(

        queues = [(queueName1)],

        containerFactory = "exclusiveListenerContainerFactory"

    )

fun handle1(...



@RabbitListener(

        queues = [(queueName2)],

        containerFactory = "exclusiveListenerContainerFactory"

    )

fun handle2(...



@Bean

open fun exclusiveListenerContainerFactory(

    connectionFactory: ConnectionFactory,

    messageConverter: MessageConverter

): SimpleRabbitListenerContainerFactory {

    val factory = SimpleRabbitListenerContainerFactory()

    factory.setConnectionFactory(connectionFactory)

    factory.setConcurrentConsumers(1)

    factory.setMessageConverter(messageConverter)

    factory.setDefaultRequeueRejected(false)

    return factory

}

I used SINGLE exclusiveListenerContainerFactory for each kind on listeners.

And SINGLE rabbitConnectionFactory (instance of CachingConnectionFactory)

I dont know is this right.

Trouble is:

Some times application hangs on Spring5. On Spring 4 same configuration work well. Funny, that it hands on machine with KVM virtualization. With Virtualbox work well even with Spring5.

jstack https://pastebin.com/MXdHcVAK

top -H -p shows PID 42 consumes all cpu

Some time ago I tried to move to Spring5 with same trouble. I thought it was reactor-netty lib issue, but no luck, they can't reproduce https://github.com/reactor/reactor-netty/issues/381

I think these two things - configuration and hanging - is bound. But I can't get deeper.

Update

Update2

Lead developer of Spring Integration and Spring AMQP - @GaryRussell - approve this configuration. So this issue is about netty I think.

edited Dec 18 '18 at 20:00

Brian Clozel

32k779104

asked Nov 22 '18 at 12:20

Lewik

597

I want to highlight: application should have multiple single threaded listeners. One listener for one queue. I mean some number of . Not multiple listeners for one queue.

I use kotlin. But the trouble is in configuration or even in library bug.

@RabbitListener(

        queues = [(queueName1)],

        containerFactory = "exclusiveListenerContainerFactory"

    )

fun handle1(...



@RabbitListener(

        queues = [(queueName2)],

        containerFactory = "exclusiveListenerContainerFactory"

    )

fun handle2(...



@Bean

open fun exclusiveListenerContainerFactory(

    connectionFactory: ConnectionFactory,

    messageConverter: MessageConverter

): SimpleRabbitListenerContainerFactory {

    val factory = SimpleRabbitListenerContainerFactory()

    factory.setConnectionFactory(connectionFactory)

    factory.setConcurrentConsumers(1)

    factory.setMessageConverter(messageConverter)

    factory.setDefaultRequeueRejected(false)

    return factory

}

I used SINGLE exclusiveListenerContainerFactory for each kind on listeners.

And SINGLE rabbitConnectionFactory (instance of CachingConnectionFactory)

I dont know is this right.

Trouble is:

Some times application hangs on Spring5. On Spring 4 same configuration work well. Funny, that it hands on machine with KVM virtualization. With Virtualbox work well even with Spring5.

jstack https://pastebin.com/MXdHcVAK

top -H -p shows PID 42 consumes all cpu

Some time ago I tried to move to Spring5 with same trouble. I thought it was reactor-netty lib issue, but no luck, they can't reproduce https://github.com/reactor/reactor-netty/issues/381

I think these two things - configuration and hanging - is bound. But I can't get deeper.

Update

Update2

Lead developer of Spring Integration and Spring AMQP - @GaryRussell - approve this configuration. So this issue is about netty I think.

netty spring-rabbitmq spring-rabbit reactor-netty

edited Dec 18 '18 at 20:00

Brian Clozel

32k779104

asked Nov 22 '18 at 12:20

Lewik

597

edited Dec 18 '18 at 20:00

Brian Clozel

32k779104

asked Nov 22 '18 at 12:20

Lewik

597

edited Dec 18 '18 at 20:00

Brian Clozel

32k779104

edited Dec 18 '18 at 20:00

Brian Clozel

32k779104

edited Dec 18 '18 at 20:00

Brian Clozel

32k779104

asked Nov 22 '18 at 12:20

Lewik

597

asked Nov 22 '18 at 12:20

Lewik

597

asked Nov 22 '18 at 12:20

Lewik

597

Take a thread dump when it hangs to see what the container threads are doing. If you can't figure it out add the dump to the question.

– Gary Russell
Nov 22 '18 at 15:32

@GaryRussell it's already here jstack pastebin.com/MXdHcVAK Is my configuration correct?

– Lewik
Nov 22 '18 at 15:43

Why do you think this is related to RabbitMQ? The container threads are all either blocked waiting for a new message or blocked in reactor code via Stomp. Looks unrelated to RabbitMQ to me.

– Gary Russell
Nov 22 '18 at 15:53

I definitely don't know source of lock. I assumed this could be rabbitmq. I thought netty used for rabbit.

– Lewik
Nov 22 '18 at 15:57

1

Yes it's fine; but concurrentConsumers defaults to 1 anyway.

– Gary Russell
Nov 22 '18 at 16:27

|
show 4 more comments

Take a thread dump when it hangs to see what the container threads are doing. If you can't figure it out add the dump to the question.

– Gary Russell
Nov 22 '18 at 15:32

@GaryRussell it's already here jstack pastebin.com/MXdHcVAK Is my configuration correct?

– Lewik
Nov 22 '18 at 15:43

Why do you think this is related to RabbitMQ? The container threads are all either blocked waiting for a new message or blocked in reactor code via Stomp. Looks unrelated to RabbitMQ to me.

– Gary Russell
Nov 22 '18 at 15:53

I definitely don't know source of lock. I assumed this could be rabbitmq. I thought netty used for rabbit.

– Lewik
Nov 22 '18 at 15:57

1

Yes it's fine; but concurrentConsumers defaults to 1 anyway.

– Gary Russell
Nov 22 '18 at 16:27

Take a thread dump when it hangs to see what the container threads are doing. If you can't figure it out add the dump to the question.

– Gary Russell
Nov 22 '18 at 15:32

@GaryRussell it's already here jstack pastebin.com/MXdHcVAK Is my configuration correct?

– Lewik
Nov 22 '18 at 15:43

Why do you think this is related to RabbitMQ? The container threads are all either blocked waiting for a new message or blocked in reactor code via Stomp. Looks unrelated to RabbitMQ to me.

– Gary Russell
Nov 22 '18 at 15:53

I definitely don't know source of lock. I assumed this could be rabbitmq. I thought netty used for rabbit.

– Lewik
Nov 22 '18 at 15:57

Yes it's fine; but concurrentConsumers defaults to 1 anyway.

– Gary Russell
Nov 22 '18 at 16:27

|
show 4 more comments

0

active

oldest

votes

Your Answer

StackExchange.ifUsing("editor", function () {
StackExchange.using("externalEditor", function () {
StackExchange.using("snippets", function () {
StackExchange.snippets.init();
});
});
}, "code-snippets");

StackExchange.ready(function() {
var channelOptions = {
tags: "".split(" "),
id: "1"
};
initTagRenderer("".split(" "), "".split(" "), channelOptions);

StackExchange.using("externalEditor", function() {
// Have to fire editor after snippets, if snippets enabled
if (StackExchange.settings.snippets.snippetsEnabled) {
StackExchange.using("snippets", function() {
createEditor();
});
}
else {
createEditor();
}
});

function createEditor() {
StackExchange.prepareEditor({
heartbeatType: 'answer',
autoActivateHeartbeat: false,
convertImagesToLinks: true,
noModals: true,
showLowRepImageUploadWarning: true,
reputationToPostImages: 10,
bindNavPrevention: true,
postfix: "",
imageUploader: {
brandingHtml: "Powered by u003ca class="icon-imgur-white" href="https://imgur.com/"u003eu003c/au003e",
contentPolicyHtml: "User contributions licensed under u003ca href="https://creativecommons.org/licenses/by-sa/3.0/"u003ecc by-sa 3.0 with attribution requiredu003c/au003e u003ca href="https://stackoverflow.com/legal/content-policy"u003e(content policy)u003c/au003e",
allowUrls: true
},
onDemand: true,
discardSelector: ".discard-answer"
,immediatelyShowMarkdownHelp:true
});

}
});

draft saved

draft discarded

Sign up or log in

StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});

Post as a guest

Name

Required, but never shown

StackExchange.ready(
function () {
StackExchange.openid.initPostLogin('.new-post-login', 'https%3a%2f%2fstackoverflow.com%2fquestions%2f53430903%2fspring-single-threaded-consumer%23new-answer', 'question_page');
}
);

Post as a guest

Name

Required, but never shown

0

active

oldest

votes

0

active

oldest

votes

draft saved

draft discarded

Thanks for contributing an answer to Stack Overflow!

Please be sure to answer the question. Provide details and share your research!

But avoid …

Asking for help, clarification, or responding to other answers.

Making statements based on opinion; back them up with references or personal experience.

To learn more, see our tips on writing great answers.

draft saved

draft discarded

Sign up or log in

StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});

Post as a guest

Name

Required, but never shown

Post as a guest

Name

Required, but never shown

Sign up or log in

StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});

Post as a guest

Name

Required, but never shown

Sign up or log in

StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});

Post as a guest

Name

Required, but never shown

Sign up or log in

StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});

Post as a guest

Name

Required, but never shown

Name

Required, but never shown

Name

Required, but never shown

This page is only for reference, If you need detailed information, please check here

搜尋此網誌

Wsrtjtyk