org.joyqueue.client.internal.consumer.support

类 DefaultMessageConsumer

java.lang.Object

org.joyqueue.toolkit.service.Activity

org.joyqueue.toolkit.service.Service

org.joyqueue.client.internal.consumer.support.DefaultMessageConsumer

所有已实现的接口:

MessageConsumer, LifeCycle, Online

public class DefaultMessageConsumer
extends Service
implements MessageConsumer

DefaultMessageConsumer author: gaohaoxiang date: 2019/1/10

嵌套类概要

从类继承的嵌套类/接口 org.joyqueue.toolkit.service.Service

Service.Election, Service.ElectionListener

从类继承的嵌套类/接口 org.joyqueue.toolkit.service.Activity

Activity.ServiceState

字段概要

字段

限定符和类型	字段和说明
protected static org.slf4j.Logger	logger

从类继承的字段 org.joyqueue.toolkit.service.Activity

readLock, rwLock, serviceState, signal, started, writeLock

构造器概要

构造器

构造器和说明
DefaultMessageConsumer(ConsumerConfig config, NameServerConfig nameServerConfig, ClusterManager clusterManager, ClusterClientManager clusterClientManager, ConsumerClientManager consumerClientManager)

方法概要

限定符和类型	方法和说明
void	addInterceptor(ConsumerInterceptor interceptor)
protected void	checkState()
protected void	checkSubscribe()
protected void	checkUnsubscribe()
org.joyqueue.exception.JoyQueueCode	commitIndex(short partition, long index)
org.joyqueue.exception.JoyQueueCode	commitMaxIndex()
org.joyqueue.exception.JoyQueueCode	commitMaxIndex(short partition)
org.joyqueue.exception.JoyQueueCode	commitMinIndex()
org.joyqueue.exception.JoyQueueCode	commitMinIndex(short partition)
protected void	doStart()
protected void	doStop()
protected void	doSubscribeListener(String topic, BaseMessageListener messageListener)
FetchIndexData	fetchIndex(short partition)
TopicMetadata	getTopicMetadata(String topic)
boolean	isListenSuspended()
boolean	isSubscribed()
protected TopicMessageConsumer	newTopicMessageConsumer(String topic)
List<ConsumeMessage>	poll()
List<ConsumeMessage>	poll(long timeout, TimeUnit timeoutUnit)
CompletableFuture<List<ConsumeMessage>>	pollAsync()
CompletableFuture<List<ConsumeMessage>>	pollAsync(long timeout, TimeUnit timeoutUnit)
ConsumeMessage	pollOnce()
ConsumeMessage	pollOnce(long timeout, TimeUnit timeoutUnit)
List<ConsumeMessage>	pollPartition(short partition)
List<ConsumeMessage>	pollPartition(short partition, long index)
List<ConsumeMessage>	pollPartition(short partition, long index, long timeout, TimeUnit timeoutUnit)
List<ConsumeMessage>	pollPartition(short partition, long timeout, TimeUnit timeoutUnit)
CompletableFuture<List<ConsumeMessage>>	pollPartitionAsync(short partition)
CompletableFuture<List<ConsumeMessage>>	pollPartitionAsync(short partition, long index)
CompletableFuture<List<ConsumeMessage>>	pollPartitionAsync(short partition, long index, long timeout, TimeUnit timeoutUnit)
CompletableFuture<List<ConsumeMessage>>	pollPartitionAsync(short partition, long timeout, TimeUnit timeoutUnit)
ConsumeMessage	pollPartitionOnce(short partition)
ConsumeMessage	pollPartitionOnce(short partition, long index)
ConsumeMessage	pollPartitionOnce(short partition, long index, long timeout, TimeUnit timeoutUnit)
ConsumeMessage	pollPartitionOnce(short partition, long timeout, TimeUnit timeoutUnit)
void	removeInterceptor(ConsumerInterceptor interceptor)
org.joyqueue.exception.JoyQueueCode	reply(List<ConsumeReply> replyList)
org.joyqueue.exception.JoyQueueCode	replyOnce(ConsumeReply reply)
void	resumeListen()
void	subscribe(String topic)
void	subscribe(String topic, MessageListener messageListener)
void	subscribeBatch(String topic, BatchMessageListener batchMessageListener)
String	subscription()
void	suspendListen()
void	unsubscribe()

从类继承的方法 org.joyqueue.toolkit.service.Service

getServiceState, isReady, isStarted, isStopped, start, stop, stop, validate, willStop

从类继承的方法 org.joyqueue.toolkit.service.Activity

afterStart, afterStop, await, beforeStart, beforeStop, convert, getReadLock, getWriteLock, startError

从类继承的方法 java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

从接口继承的方法 org.joyqueue.toolkit.lang.LifeCycle

start, stop

从接口继承的方法 org.joyqueue.toolkit.lang.Online

isStarted

字段详细资料

logger

protected static final org.slf4j.Logger logger

构造器详细资料

DefaultMessageConsumer

public DefaultMessageConsumer(ConsumerConfig config,
                              NameServerConfig nameServerConfig,
                              ClusterManager clusterManager,
                              ClusterClientManager clusterClientManager,
                              ConsumerClientManager consumerClientManager)

方法详细资料

doStart

protected void doStart()
                throws Exception

覆盖:

doStart 在类中 Activity

抛出:

Exception

doStop

protected void doStop()

覆盖:

doStop 在类中 Activity

subscribe

public void subscribe(String topic)

指定者:

subscribe 在接口中 MessageConsumer

unsubscribe

public void unsubscribe()

指定者:

unsubscribe 在接口中 MessageConsumer

subscription

public String subscription()

指定者:

subscription 在接口中 MessageConsumer

isSubscribed

public boolean isSubscribed()

指定者:

isSubscribed 在接口中 MessageConsumer

subscribe

public void subscribe(String topic,
                      MessageListener messageListener)

指定者:

subscribe 在接口中 MessageConsumer

subscribeBatch

public void subscribeBatch(String topic,
                           BatchMessageListener batchMessageListener)

指定者:

subscribeBatch 在接口中 MessageConsumer

doSubscribeListener

protected void doSubscribeListener(String topic,
                                   BaseMessageListener messageListener)

newTopicMessageConsumer

protected TopicMessageConsumer newTopicMessageConsumer(String topic)

resumeListen

public void resumeListen()

指定者:

resumeListen 在接口中 MessageConsumer

suspendListen

public void suspendListen()

指定者:

suspendListen 在接口中 MessageConsumer

isListenSuspended

public boolean isListenSuspended()

指定者:

isListenSuspended 在接口中 MessageConsumer

addInterceptor

public void addInterceptor(ConsumerInterceptor interceptor)

指定者:

addInterceptor 在接口中 MessageConsumer

removeInterceptor

public void removeInterceptor(ConsumerInterceptor interceptor)

指定者:

removeInterceptor 在接口中 MessageConsumer

pollOnce

public ConsumeMessage pollOnce()

指定者:

pollOnce 在接口中 MessageConsumer

pollOnce

public ConsumeMessage pollOnce(long timeout,
                               TimeUnit timeoutUnit)

指定者:

pollOnce 在接口中 MessageConsumer

poll

public List<ConsumeMessage> poll()

指定者:

poll 在接口中 MessageConsumer

poll

public List<ConsumeMessage> poll(long timeout,
                                 TimeUnit timeoutUnit)

指定者:

poll 在接口中 MessageConsumer

pollAsync

public CompletableFuture<List<ConsumeMessage>> pollAsync()

指定者:

pollAsync 在接口中 MessageConsumer

pollAsync

public CompletableFuture<List<ConsumeMessage>> pollAsync(long timeout,
                                                         TimeUnit timeoutUnit)

指定者:

pollAsync 在接口中 MessageConsumer

pollPartitionOnce

public ConsumeMessage pollPartitionOnce(short partition)

指定者:

pollPartitionOnce 在接口中 MessageConsumer

pollPartitionOnce

public ConsumeMessage pollPartitionOnce(short partition,
                                        long timeout,
                                        TimeUnit timeoutUnit)

指定者:

pollPartitionOnce 在接口中 MessageConsumer

pollPartitionOnce

public ConsumeMessage pollPartitionOnce(short partition,
                                        long index)

指定者:

pollPartitionOnce 在接口中 MessageConsumer

pollPartitionOnce

public ConsumeMessage pollPartitionOnce(short partition,
                                        long index,
                                        long timeout,
                                        TimeUnit timeoutUnit)

指定者:

pollPartitionOnce 在接口中 MessageConsumer

pollPartition

public List<ConsumeMessage> pollPartition(short partition)

指定者:

pollPartition 在接口中 MessageConsumer

pollPartition

public List<ConsumeMessage> pollPartition(short partition,
                                          long timeout,
                                          TimeUnit timeoutUnit)

指定者:

pollPartition 在接口中 MessageConsumer

pollPartitionAsync

public CompletableFuture<List<ConsumeMessage>> pollPartitionAsync(short partition)

指定者:

pollPartitionAsync 在接口中 MessageConsumer

pollPartitionAsync

public CompletableFuture<List<ConsumeMessage>> pollPartitionAsync(short partition,
                                                                  long timeout,
                                                                  TimeUnit timeoutUnit)

指定者:

pollPartitionAsync 在接口中 MessageConsumer

pollPartition

public List<ConsumeMessage> pollPartition(short partition,
                                          long index)

指定者:

pollPartition 在接口中 MessageConsumer

pollPartition

public List<ConsumeMessage> pollPartition(short partition,
                                          long index,
                                          long timeout,
                                          TimeUnit timeoutUnit)

指定者:

pollPartition 在接口中 MessageConsumer

pollPartitionAsync

public CompletableFuture<List<ConsumeMessage>> pollPartitionAsync(short partition,
                                                                  long index)

指定者:

pollPartitionAsync 在接口中 MessageConsumer

pollPartitionAsync

public CompletableFuture<List<ConsumeMessage>> pollPartitionAsync(short partition,
                                                                  long index,
                                                                  long timeout,
                                                                  TimeUnit timeoutUnit)

指定者:

pollPartitionAsync 在接口中 MessageConsumer

reply

public org.joyqueue.exception.JoyQueueCode reply(List<ConsumeReply> replyList)

指定者:

reply 在接口中 MessageConsumer

replyOnce

public org.joyqueue.exception.JoyQueueCode replyOnce(ConsumeReply reply)

指定者:

replyOnce 在接口中 MessageConsumer

commitIndex

public org.joyqueue.exception.JoyQueueCode commitIndex(short partition,
                                                       long index)

指定者:

commitIndex 在接口中 MessageConsumer

commitMaxIndex

public org.joyqueue.exception.JoyQueueCode commitMaxIndex()

指定者:

commitMaxIndex 在接口中 MessageConsumer

commitMaxIndex

public org.joyqueue.exception.JoyQueueCode commitMaxIndex(short partition)

指定者:

commitMaxIndex 在接口中 MessageConsumer

commitMinIndex

public org.joyqueue.exception.JoyQueueCode commitMinIndex()

指定者:

commitMinIndex 在接口中 MessageConsumer

commitMinIndex

public org.joyqueue.exception.JoyQueueCode commitMinIndex(short partition)

指定者:

commitMinIndex 在接口中 MessageConsumer

fetchIndex

public FetchIndexData fetchIndex(short partition)

指定者:

fetchIndex 在接口中 MessageConsumer

getTopicMetadata

public TopicMetadata getTopicMetadata(String topic)

指定者:

getTopicMetadata 在接口中 MessageConsumer

checkState

protected void checkState()

checkSubscribe

protected void checkSubscribe()

checkUnsubscribe

protected void checkUnsubscribe()