subscription

Documentation

Overview

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.

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.

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.

Index

• type SubscriptionGroupConfig
  • func NewSubscriptionGroupConfig() *SubscriptionGroupConfig
  • func (sgc *SubscriptionGroupConfig) String() string
• type SubscriptionGroupTable
  • func NewSubscriptionGroupTable() *SubscriptionGroupTable
  • func (table *SubscriptionGroupTable) Clear()
  • func (table *SubscriptionGroupTable) ClearAndPutAll(offsetMap *concurrent.ConcurrentMap)
  • func (table *SubscriptionGroupTable) Foreach(fn func(k string, v *SubscriptionGroupConfig))
  • func (table *SubscriptionGroupTable) Get(k string) *SubscriptionGroupConfig
  • func (table *SubscriptionGroupTable) Put(k string, v *SubscriptionGroupConfig) *SubscriptionGroupConfig
  • func (table *SubscriptionGroupTable) PutAll(offsetMap *concurrent.ConcurrentMap)
  • func (table *SubscriptionGroupTable) Remove(k string) *SubscriptionGroupConfig
  • func (table *SubscriptionGroupTable) Size() int
• type SubscriptionGroupWrapper
  • func NewSubscriptionGroupWrapper() *SubscriptionGroupWrapper

Types

type SubscriptionGroupConfig

type SubscriptionGroupConfig struct {
	GroupName                    string `json:"groupName"`                    // 订阅组名
	ConsumeEnable                bool   `json:"consumeEnable"`                // 消费功能是否开启
	ConsumeFromMinEnable         bool   `json:"consumeFromMinEnable"`         // 是否允许从队列最小位置开始消费(线上默认会设置为false)
	ConsumeBroadcastEnable       bool   `json:"consumeBroadcastEnable"`       // 是否允许广播方式消费
	RetryQueueNums               int32  `json:"retryQueueNums"`               // 每个订阅组配置几个重试队列(消费失败的消息放到一个重试队列)
	RetryMaxTimes                int32  `json:"retryMaxTimes"`                // 重试消费最大次数(超过最大次数，则投递到死信队列并且不再投递，并报警)
	BrokerId                     int64  `json:"brokerId"`                     // 从哪个Broker开始消费
	WhichBrokerWhenConsumeSlowly int64  `json:"whichBrokerWhenConsumeSlowly"` // 发现消息堆积后，将Consumer的消费请求重定向到另外一台Slave机器
}

SubscriptionGroupConfig 订阅关系配置 Author gaoyanlei Since 2017/8/9

func NewSubscriptionGroupConfig

func NewSubscriptionGroupConfig() *SubscriptionGroupConfig

NewSubscriptionGroupConfig 初始化SubscriptionGroupConfig Author gaoyanlei Since 2017/8/9

func (*SubscriptionGroupConfig) String

func (sgc *SubscriptionGroupConfig) String() string

String 打印SubscriptionGroupConfig结构体数据 Author: tianyuliang Since: 2017/10/13

type SubscriptionGroupTable

type SubscriptionGroupTable struct {
	Tables      map[string]*SubscriptionGroupConfig `json:"tables"` // key:group
	DataVersion basis.DataVersion                   `json:"dataVersion"`
	// contains filtered or unexported fields
}

func NewSubscriptionGroupTable

func NewSubscriptionGroupTable() *SubscriptionGroupTable

func (*SubscriptionGroupTable) Clear

func (table *SubscriptionGroupTable) Clear()

Clear 清空 Author rongzhihong Since 2017/9/18

func (*SubscriptionGroupTable) ClearAndPutAll

func (table *SubscriptionGroupTable) ClearAndPutAll(offsetMap *concurrent.ConcurrentMap)

ClearAndPutAll 清空map,再PutAll Author rongzhihong Since 2017/9/18

func (*SubscriptionGroupTable) Foreach

func (table *SubscriptionGroupTable) Foreach(fn func(k string, v *SubscriptionGroupConfig))

func (*SubscriptionGroupTable) Get

func (table *SubscriptionGroupTable) Get(k string) *SubscriptionGroupConfig

func (*SubscriptionGroupTable) Put

func (table *SubscriptionGroupTable) Put(k string, v *SubscriptionGroupConfig) *SubscriptionGroupConfig

func (*SubscriptionGroupTable) PutAll

func (table *SubscriptionGroupTable) PutAll(offsetMap *concurrent.ConcurrentMap)

syncTopicConfig 同步Topic配置文件 Author rongzhihong Since 2017/9/18

func (*SubscriptionGroupTable) Remove

func (table *SubscriptionGroupTable) Remove(k string) *SubscriptionGroupConfig

func (*SubscriptionGroupTable) Size

func (table *SubscriptionGroupTable) Size() int

type SubscriptionGroupWrapper

type SubscriptionGroupWrapper struct {
	SubscriptionGroupTable *concurrent.ConcurrentMap `json:"subscriptionGroupTable"`
	DataVersion            basis.DataVersion         `json:"dataVersion"`
}

SubscriptionGroupWrapper 订阅组配置，序列化包装 Author gaoyanlei Since 2017/8/22

func NewSubscriptionGroupWrapper

func NewSubscriptionGroupWrapper() *SubscriptionGroupWrapper