rabbitmq

README

RabbitMQ

Installation

go get -u github.com/gojuukaze/YTask/drives/rabbitmq/v3

Broker

package main

import (
    "github.com/gojuukaze/YTask/drives/rabbitmq/v3"
)

func main() {
	broker := rabbitmq.NewRabbitMqBroker("127.0.0.1", "5672", "guest", "guest", "", 2)
	// ...
}

Documentation

Index

• Variables
• type Broker
  • func NewRabbitMqBroker(host, port, user, password, vhost string, poolSize int) Broker
  • func (r *Broker) Activate()
  • func (r Broker) Clone() brokers.BrokerInterface
  • func (r *Broker) GetPoolSize() int
  • func (r *Broker) LSend(queueName string, msg message.Message) error
  • func (r *Broker) Next(queueName string) (message.Message, error)
  • func (r *Broker) Send(queueName string, msg message.Message) error
  • func (r *Broker) SetPoolSize(n int)
• type Client
  • func NewRabbitMqClient(host, port, user, password, vhost string, maxChannel int) *Client
  • func (c *Client) Get(queueName string) ([]byte, error)
  • func (c *Client) GetChannel() (*amqp.Channel, error)
  • func (c *Client) Publish(queueName string, value []byte, Priority uint8) error
  • func (c *Client) PutChannel(channel *amqp.Channel, isBad bool)
  • func (c *Client) QueueDeclare(queueName string, channel *amqp.Channel) error

Variables

var (
	AMQPNil          = errors.New("rabbitMq get nil")
	ErrNoIdleChannel = errors.New("rabbitMq no idle channel")
	GetChanTimeout   = 60 * time.Second
	GetChanSleepTime = 100 * time.Millisecond
)

Types

type Broker

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

func NewRabbitMqBroker

func NewRabbitMqBroker(host, port, user, password, vhost string, poolSize int) Broker

func (*Broker) Activate

func (r *Broker) Activate()

func (Broker) Clone

func (r Broker) Clone() brokers.BrokerInterface

func (*Broker) GetPoolSize

func (r *Broker) GetPoolSize() int

func (*Broker) LSend

func (r *Broker) LSend(queueName string, msg message.Message) error

func (*Broker) Next

func (r *Broker) Next(queueName string) (message.Message, error)

func (*Broker) Send

func (r *Broker) Send(queueName string, msg message.Message) error

func (*Broker) SetPoolSize

func (r *Broker) SetPoolSize(n int)

type Client

type Client struct {
	Conn       *amqp.Connection
	IdleChan   map[string]*amqp.Channel
	NumOpen    int
	MaxChannel int
	// contains filtered or unexported fields
}

func NewRabbitMqClient

func NewRabbitMqClient(host, port, user, password, vhost string, maxChannel int) *Client

func (*Client) Get

func (c *Client) Get(queueName string) ([]byte, error)

func (*Client) GetChannel

func (c *Client) GetChannel() (*amqp.Channel, error)

func (*Client) Publish

func (c *Client) Publish(queueName string, value []byte, Priority uint8) error

func (*Client) PutChannel

func (c *Client) PutChannel(channel *amqp.Channel, isBad bool)

func (*Client) QueueDeclare

func (c *Client) QueueDeclare(queueName string, channel *amqp.Channel) error

创建队列，如果队列已经存在，则忽略