com.taobao.metamorphosis.client.consumer
类 SimpleMessageConsumer

java.lang.Object
  com.taobao.metamorphosis.client.consumer.SimpleMessageConsumer

所有已实现的接口：

InnerConsumer, MessageConsumer, Shutdownable

public class SimpleMessageConsumer
extends Object
implements MessageConsumer, InnerConsumer

消息消费者基类

作者：

boyan, wuhua

嵌套类摘要

static class	SimpleMessageConsumer.DropPolicy Created with IntelliJ IDEA.
static class	SimpleMessageConsumer.LocalRecoverPolicy Created with IntelliJ IDEA.

构造方法摘要

SimpleMessageConsumer(MetaMessageSessionFactory messageSessionFactory, RemotingClientWrapper remotingClient, ConsumerConfig consumerConfig, ConsumerZooKeeper consumerZooKeeper, ProducerZooKeeper producerZooKeeper, SubscribeInfoManager subscribeInfoManager, RecoverManager recoverManager, OffsetStorage offsetStorage, LoadBalanceStrategy loadBalanceStrategy)

方法摘要

void	appendCouldNotProcessMessage(Message message) 处理无法被客户端消费的消息
void	completeSubscribe() 使得已经订阅的topic生效,此方法仅能调用一次,再次调用无效并将抛出异常
MessageIterator	fetch(FetchRequest fetchRequest, long timeout, TimeUnit timeUnit) 抓取消息
MessageIterator	get(String topic, Partition partition, long offset, int maxSize) 获取指定topic和分区下面的消息，默认超时10秒
MessageIterator	get(String topic, Partition partition, long offset, int maxSize, long timeout, TimeUnit timeUnit) 获取指定topic和分区下面的消息，在指定时间内没有返回则抛出异常
ConsumerConfig	getConsumerConfig() 返回消费者配置
FetchManager	getFetchManager()
ConsumerMessageFilter	getMessageFilter(String topic)
MessageListener	getMessageListener(String topic) 返回topic对应的消息监听器
OffsetStorage	getOffsetStorage() 返回此消费者使用的offset存储器，可共享给其他消费者
MetaMessageSessionFactory	getParent()
RejectConsumptionHandler	getRejectConsumptionHandler() Returns current RejectConsumptionHandler
long	offset(FetchRequest fetchRequest) 查询offset
void	setRejectConsumptionHandler(RejectConsumptionHandler rejectConsumptionHandler) Sets RejectConsumptionHandler for this consumer.
void	setSubscriptions(Collection<com.taobao.metamorphosis.client.consumer.Subscription> subscriptions) 批量订阅消息,请注意，调用此方法并不会使订阅关系立即生效，只有在调用complete方法后才生效。
void	shutdown() 停止消费者
MessageConsumer	subscribe(String topic, int maxSize, MessageListener messageListener) 订阅指定的消息，传入MessageListener，当有消息达到的时候主动通知MessageListener，请注意， 调用此方法并不会使订阅关系立即生效， 只有在调用complete方法后才生效，此方法可做链式调用
MessageConsumer	subscribe(String topic, int maxSize, MessageListener messageListener, ConsumerMessageFilter filter) 订阅指定的消息，传入MessageListener和ConsumerMessageFilter， 当有消息到达并且ConsumerMessageFilter #accept返回true的时候,主动通知MessageListener该条消息，请注意， 调用此方法并不会使订阅关系立即生效， 只有在调用complete方法后才生效，此方法可做链式调用

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

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

构造方法详细信息

SimpleMessageConsumer

public SimpleMessageConsumer(MetaMessageSessionFactory messageSessionFactory,
                             RemotingClientWrapper remotingClient,
                             ConsumerConfig consumerConfig,
                             ConsumerZooKeeper consumerZooKeeper,
                             ProducerZooKeeper producerZooKeeper,
                             SubscribeInfoManager subscribeInfoManager,
                             RecoverManager recoverManager,
                             OffsetStorage offsetStorage,
                             LoadBalanceStrategy loadBalanceStrategy)

方法详细信息

getParent

public MetaMessageSessionFactory getParent()

getFetchManager

public FetchManager getFetchManager()

getOffsetStorage

public OffsetStorage getOffsetStorage()

从接口 MessageConsumer 复制的描述

返回此消费者使用的offset存储器，可共享给其他消费者

指定者：

接口 MessageConsumer 中的 getOffsetStorage

返回：

shutdown

public void shutdown()
              throws MetaClientException

从接口 MessageConsumer 复制的描述

停止消费者

指定者：

接口 MessageConsumer 中的 shutdown

指定者：

接口 Shutdownable 中的 shutdown

抛出：

MetaClientException

subscribe

public MessageConsumer subscribe(String topic,
                                 int maxSize,
                                 MessageListener messageListener)
                          throws MetaClientException

从接口 MessageConsumer 复制的描述

订阅指定的消息，传入MessageListener，当有消息达到的时候主动通知MessageListener，请注意， 调用此方法并不会使订阅关系立即生效， 只有在调用complete方法后才生效，此方法可做链式调用

指定者：

接口 MessageConsumer 中的 subscribe

参数：

topic - 订阅的topic

maxSize - 订阅每次接收的最大数据大小

messageListener - 消息监听器

抛出：

MetaClientException

subscribe

public MessageConsumer subscribe(String topic,
                                 int maxSize,
                                 MessageListener messageListener,
                                 ConsumerMessageFilter filter)
                          throws MetaClientException

从接口 MessageConsumer 复制的描述

订阅指定的消息，传入MessageListener和ConsumerMessageFilter， 当有消息到达并且ConsumerMessageFilter #accept返回true的时候,主动通知MessageListener该条消息，请注意， 调用此方法并不会使订阅关系立即生效， 只有在调用complete方法后才生效，此方法可做链式调用

指定者：

接口 MessageConsumer 中的 subscribe

参数：

topic - 订阅的topic

maxSize - 订阅每次接收的最大数据大小

抛出：

MetaClientException

appendCouldNotProcessMessage

public void appendCouldNotProcessMessage(Message message)
                                  throws IOException

从接口 InnerConsumer 复制的描述

处理无法被客户端消费的消息

指定者：

接口 InnerConsumer 中的 appendCouldNotProcessMessage

抛出：

IOException

completeSubscribe

public void completeSubscribe()
                       throws MetaClientException

从接口 MessageConsumer 复制的描述

使得已经订阅的topic生效,此方法仅能调用一次,再次调用无效并将抛出异常

指定者：

接口 MessageConsumer 中的 completeSubscribe

抛出：

MetaClientException

getMessageListener

public MessageListener getMessageListener(String topic)

从接口 InnerConsumer 复制的描述

返回topic对应的消息监听器

指定者：

接口 InnerConsumer 中的 getMessageListener

返回：

getMessageFilter

public ConsumerMessageFilter getMessageFilter(String topic)

指定者：

接口 InnerConsumer 中的 getMessageFilter

offset

public long offset(FetchRequest fetchRequest)
            throws MetaClientException

从接口 InnerConsumer 复制的描述

查询offset

指定者：

接口 InnerConsumer 中的 offset

返回：

抛出：

MetaClientException

fetch

public MessageIterator fetch(FetchRequest fetchRequest,
                             long timeout,
                             TimeUnit timeUnit)
                      throws MetaClientException,
                             InterruptedException

从接口 InnerConsumer 复制的描述

抓取消息

指定者：

接口 InnerConsumer 中的 fetch

返回：

抛出：

MetaClientException

InterruptedException

setSubscriptions

public void setSubscriptions(Collection<com.taobao.metamorphosis.client.consumer.Subscription> subscriptions)
                      throws MetaClientException

从接口 MessageConsumer 复制的描述

批量订阅消息,请注意，调用此方法并不会使订阅关系立即生效，只有在调用complete方法后才生效。

指定者：

接口 MessageConsumer 中的 setSubscriptions

抛出：

MetaClientException

get

public MessageIterator get(String topic,
                           Partition partition,
                           long offset,
                           int maxSize,
                           long timeout,
                           TimeUnit timeUnit)
                    throws MetaClientException,
                           InterruptedException

从接口 MessageConsumer 复制的描述

获取指定topic和分区下面的消息，在指定时间内没有返回则抛出异常

指定者：

接口 MessageConsumer 中的 get

返回：

消息迭代器，可能为null

抛出：

MetaClientException

InterruptedException

getRejectConsumptionHandler

public RejectConsumptionHandler getRejectConsumptionHandler()

从接口 MessageConsumer 复制的描述

Returns current RejectConsumptionHandler

指定者：

接口 MessageConsumer 中的 getRejectConsumptionHandler

返回：

setRejectConsumptionHandler

public void setRejectConsumptionHandler(RejectConsumptionHandler rejectConsumptionHandler)

从接口 MessageConsumer 复制的描述

Sets RejectConsumptionHandler for this consumer.

指定者：

接口 MessageConsumer 中的 setRejectConsumptionHandler

getConsumerConfig

public ConsumerConfig getConsumerConfig()

从接口 MessageConsumer 复制的描述

返回消费者配置

指定者：

接口 InnerConsumer 中的 getConsumerConfig

指定者：

接口 MessageConsumer 中的 getConsumerConfig

get

public MessageIterator get(String topic,
                           Partition partition,
                           long offset,
                           int maxSize)
                    throws MetaClientException,
                           InterruptedException

从接口 MessageConsumer 复制的描述

获取指定topic和分区下面的消息，默认超时10秒

指定者：

接口 MessageConsumer 中的 get

返回：

消息迭代器，可能为null

抛出：

MetaClientException

InterruptedException