From cd8780e82afcd106eefab4fedfe8be4471611574 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Tue, 23 Aug 2022 16:29:35 -0400 Subject: [PATCH] Initial commit for concurrent message listeners --- ...bstractPulsarMessageListenerContainer.java | 40 +++++- ...currentPulsarMessageListenerContainer.java | 128 ++++++++++++++++++ ...DefaultPulsarMessageListenerContainer.java | 19 +-- 3 files changed, 169 insertions(+), 18 deletions(-) create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainer.java diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java index 0885393b..5fc3b915 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java @@ -27,6 +27,7 @@ import org.springframework.context.ApplicationEventPublisherAware; import org.springframework.core.log.LogAccessor; import org.springframework.lang.Nullable; import org.springframework.pulsar.core.PulsarConsumerFactory; +import org.springframework.util.Assert; /** * Base implementation for the {@link PulsarMessageListenerContainer}. @@ -47,12 +48,16 @@ public abstract class AbstractPulsarMessageListenerContainer implements Pulsa private final PulsarContainerProperties pulsarContainerProperties; - private final PulsarConsumerFactory pulsarConsumerFactory; + protected final PulsarConsumerFactory pulsarConsumerFactory; private boolean autoStartup = true; private int phase; + protected final Object lifecycleMonitor = new Object(); + + private volatile boolean running = false; + @SuppressWarnings("unchecked") protected AbstractPulsarMessageListenerContainer(PulsarConsumerFactory pulsarConsumerFactory, PulsarContainerProperties pulsarContainerProperties) { @@ -66,6 +71,15 @@ public abstract class AbstractPulsarMessageListenerContainer implements Pulsa this.applicationEventPublisher = applicationEventPublisher; } + @Override + public boolean isRunning() { + return this.running; + } + + protected void setRunning(boolean running) { + this.running = running; + } + /** * Get the event publisher. * @return the publisher @@ -135,4 +149,28 @@ public abstract class AbstractPulsarMessageListenerContainer implements Pulsa return this.pulsarContainerProperties; } + protected abstract void doStart(); + + protected abstract void doStop(); + + @Override + public final void start() { + synchronized (this.lifecycleMonitor) { + if (!isRunning()) { + Assert.state(this.pulsarContainerProperties.getMessageListener() instanceof PulsarRecordMessageListener, + () -> "A " + PulsarRecordMessageListener.class.getName() + " implementation must be provided"); + doStart(); + } + } + } + + @Override + public void stop() { + synchronized (this.lifecycleMonitor) { + if (!isRunning()) { + doStop(); + } + } + } + } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainer.java new file mode 100644 index 00000000..a5722973 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainer.java @@ -0,0 +1,128 @@ +/* + * Copyright 2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.pulsar.listener; + +import java.util.ArrayList; +import java.util.List; + +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.core.task.AsyncTaskExecutor; +import org.springframework.core.task.SimpleAsyncTaskExecutor; +import org.springframework.pulsar.core.PulsarConsumerFactory; +import org.springframework.util.Assert; + +/** + * Creates a concurrent execution context of {@link DefaultPulsarMessageListenerContainer} + * instances based on the {@link #setConcurrency(int) concurrency}. Concurrency > 1 is not + * allowed for exclusive subscriptions. + * + * @param the payload type. + * @author Soby Chacko + */ +public class ConcurrentPulsarMessageListenerContainer extends AbstractPulsarMessageListenerContainer { + + private final List> containers = new ArrayList<>(); + + private int concurrency = 1; + + private final List executors = new ArrayList<>(); + + protected ConcurrentPulsarMessageListenerContainer(PulsarConsumerFactory pulsarConsumerFactory, + PulsarContainerProperties pulsarContainerProperties) { + super(pulsarConsumerFactory, pulsarContainerProperties); + } + + public int getConcurrency() { + return this.concurrency; + } + + /** + * The maximum number of concurrent {@link DefaultPulsarMessageListenerContainer}s + * running. Messages from within the same partition will be processed sequentially. + * Concurrency > 1 is not allowed for exclusive subscriptions. + * @param concurrency the concurrency. + */ + public void setConcurrency(int concurrency) { + Assert.isTrue(concurrency > 0, "concurrency must be greater than 0"); + this.concurrency = concurrency; + } + + @Override + public void doStart() { + if (!isRunning()) { + PulsarContainerProperties containerProperties = getContainerProperties(); + + setRunning(true); + + for (int i = 0; i < this.concurrency; i++) { + DefaultPulsarMessageListenerContainer container = constructContainer(containerProperties); + configureChildContainer(i, container); + + container.start(); + this.containers.add(container); + } + } + } + + private DefaultPulsarMessageListenerContainer constructContainer(PulsarContainerProperties containerProperties) { + return new DefaultPulsarMessageListenerContainer<>(this.pulsarConsumerFactory, containerProperties); + } + + private void configureChildContainer(int index, DefaultPulsarMessageListenerContainer container) { + String beanName = getBeanName(); + beanName = (beanName == null ? "consumer" : beanName) + "-" + index; + container.setBeanName(beanName); + ApplicationContext applicationContext = getApplicationContext(); + if (applicationContext != null) { + container.setApplicationContext(applicationContext); + } + ApplicationEventPublisher publisher = getApplicationEventPublisher(); + if (publisher != null) { + container.setApplicationEventPublisher(publisher); + } + + AsyncTaskExecutor exec = container.getContainerProperties().getConsumerTaskExecutor(); + if (exec == null) { + if ((this.executors.size() > index)) { + exec = this.executors.get(index); + } + else { + exec = new SimpleAsyncTaskExecutor(beanName + "-C-"); + this.executors.add(exec); + } + container.getContainerProperties().setConsumerTaskExecutor(exec); + } + } + + @Override + public void doStop() { + if (isRunning()) { + setRunning(false); + for (DefaultPulsarMessageListenerContainer pulsarMessageListenerContainer : this.containers) { + pulsarMessageListenerContainer.stop(); + } + this.containers.clear(); + } + } + + @Override + public boolean isRunning() { + return false; + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java index 5fcada54..7f58355d 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java @@ -59,8 +59,6 @@ import org.springframework.util.StringUtils; */ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMessageListenerContainer { - private volatile boolean running = false; - private volatile CompletableFuture listenerConsumerFuture; private volatile Listener listenerConsumer; @@ -76,11 +74,7 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess } @Override - public void start() { - doStart(); - } - - private void doStart() { + protected void doStart() { PulsarContainerProperties containerProperties = getPulsarContainerProperties(); @@ -114,7 +108,7 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess } @Override - public void stop() { + public void doStop() { setRunning(false); this.logger.info("Pausing this consumer."); this.listenerConsumer.consumer.pause(); @@ -127,15 +121,6 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess } } - @Override - public boolean isRunning() { - return this.running; - } - - protected void setRunning(boolean running) { - this.running = running; - } - @Override public void destroy() {