kafka

Documentation

Index

• func NewMQ() mq.IReactMQ
• type ReactKafka
  • func (this *ReactKafka) Close()
  • func (this *ReactKafka) Connect(mode mq.Mode, strURL string) (err error)
  • func (this *ReactKafka) Consume(strBindingKey, strQueueName string, handler mq.ReactHandler) (err error)
  • func (this *ReactKafka) Debug(enable bool)
  • func (this *ReactKafka) GetAdapter() (adapter mq.Adapter)
  • func (this *ReactKafka) IsClosed() bool
  • func (this *ReactKafka) NewTLSConfig(ca, key, crt string) *tls.Config
  • func (this *ReactKafka) OnConnect()
  • func (this *ReactKafka) OnDisconnect(err error)
  • func (this *ReactKafka) OnReceive()
  • func (this *ReactKafka) Publish(strBindingKey, strQueueName, key string, value string) (err error)
  • func (this *ReactKafka) Reconnect() (err error)

Functions

func NewMQ

func NewMQ() mq.IReactMQ

Types

type ReactKafka

type ReactKafka struct {
	// contains filtered or unexported fields
}

func (*ReactKafka) Close

func (this *ReactKafka) Close()

关闭MQ

func (*ReactKafka) Connect

func (this *ReactKafka) Connect(mode mq.Mode, strURL string) (err error)

* @brief MQ服务器连接接口定义 * @param strUrl 连接服务器URL(格式规范 kafka://127.0.0.1:9092) * @return err 连接失败返回具体错误信息

func (*ReactKafka) Consume

func (this *ReactKafka) Consume(strBindingKey, strQueueName string, handler mq.ReactHandler) (err error)

* @brief 消息消费接口定义 * @param strBindingKey 队列绑定Key(topic) * @param strQueueName 队列名称(group) * @param handler 消费回调处理对象 * @return err 成功返回nil，失败返回返回具体错误信息 * @remark 服务器异常或重启时内部会自动重连服务器

func (*ReactKafka) Debug

func (this *ReactKafka) Debug(enable bool)

* @brief 开启或关闭调式模式 * @param enable true开启/false关闭

func (*ReactKafka) GetAdapter

func (this *ReactKafka) GetAdapter() (adapter mq.Adapter)

* @brief 获取当前MQ类型 * @param adapter MQ类型

func (*ReactKafka) IsClosed

func (this *ReactKafka) IsClosed() bool

* @brief 判定是否MQ服务器断开连接（异常宕机或重启） * @param * @return 远程服务器连接断开返回true，否则返回false

func (*ReactKafka) NewTLSConfig

func (this *ReactKafka) NewTLSConfig(ca, key, crt string) *tls.Config

func (*ReactKafka) OnConnect

func (this *ReactKafka) OnConnect()

MQTT 服务器连接成功回调

func (*ReactKafka) OnDisconnect

func (this *ReactKafka) OnDisconnect(err error)

MQTT 服务器断开连接回调

func (*ReactKafka) OnReceive

func (this *ReactKafka) OnReceive()

MQTT 服务器接收数据回调

func (*ReactKafka) Publish

func (this *ReactKafka) Publish(strBindingKey, strQueueName, key string, value string) (err error)

* @brief 消息发布接口定义(仅支持字符串类型消息) * @param strBindingKey 队列绑定Key(topic) * @param strQueueName 队列名称(group) * @param key 消息KEY(仅kafka必填，其他MQ类型默认填PRODUCER_KEY_NULL) * @param value 消息数据 * @return err 发布失败返回具体错误信息

func (*ReactKafka) Reconnect

func (this *ReactKafka) Reconnect() (err error)

* @brief MQ服务器重新连接接口定义 * @param * @return err 连接失败返回具体错误信息 * @remark 当Publish返回错误且IsClosed()方法亦返回true时调用此方法重连MQ服务器 * 如果已使用Consume订阅过，内部会监听连接断开事件并自动重连。