rabbit入门教程

标签：情况   local   channel   rpc   图片   规则   rect   code   product   

简介

rabbitmq是一个消息代理系统，为应用提供一个通用得消息发布，接受平台，为应用提供非阻塞的消息系统，方便进行异步处理。

优点

1. 消息的可靠性。持久化消息，消息接受确认，消息重传等可靠机制。
2. 灵活的路由。交换机可以根据广播，或者根据路由键或匹配符匹配到不同的队列。
3. 高可用的集群。

应用场景

1.异步处理

减少用户对不必要的耗时操作的等待，处理结果以异步方式（邮件，消息推送）进行提醒。

2.应用解耦

当某个应用发展到一定规模的时候，需要把里面的模块分别拆出来进行解耦，而模块之间的通讯方式是多样的，常见的有rpc，消息队列，http请求。其中消息队列在内部模块通信是更为稳定。

3.流量削峰

如果突发遇到大量的数据请求的时候，服务器如果不做队列处理，一下子处理全部的请求，会很容易造成宕机，如果把请求的数据都放入队列里，之后再逐个逐个地进行处理，可以平缓地渡过流量高峰期。

工作方式

rabbitmq的工作方式如下，生产者（publisher）发送消息到交换机，交换机（exchange）根据自己的类型以及消息的路由键，路由到对应的队列里，队列分发消息到消费者（consumer）

初尝rabbitmq

现在我们假设有这个场景，客服A需要发送客户的下单信息给库存人员B，客服A有一个订单信息发送器，库存人员B拥有消息接收器。
首先库存人员B建立连接并接受消息，伪代码：

// 建立连接
conn, _ := amqp.Dial("amqp://localhost")
ch, _ := conn.Channel()
// 声明队列，不存在则创建，存在则不会进行任何操作
queue, _ := ch.QueueDeclare("order")
// 从队列里面获取消息
deliver, _ := ch.Consume(q.Name)
for d:= range deliver {
    // 输出消息主体
    log.Printf("B Received a message: %s", d.Body)
    // 返回获取成功标识给队列
    d.Ack(true)
}

然后客服A也建立连接并发送消息，伪代码：

// 建立连接
conn, _ := amqp.Dial("amqp://localhost")
ch, _ := conn.Chanenel()
// 声明队列，不存在则创建，存在则不会进行任何操作
queue, _ := ch.QueueDeclare("order")
// 发布消息
ch.Publish(
    q.Name, // 队列名字
    amqp.Publishing{
        Body:[]byte("new order" + product.String()),
    })

上面就是一种简单的直接通过队列进行连接的方法，可能会有人看出来，为什么没有交换机的参与，其实上面的操作其实是通过默认交换机进行消息传递，可以不指定交换机名字直接指定队列名字进行交互。

模块介绍

通过上面的简单例子，我们可以更进一步地了解到rabbitmq的工作方式，下面我会更详细地讲解各个模块。

消息

消息是通信内容的主体，消息对象有点像http的request，除了可以携带消息内容，还可以带有各种属性，如：

1. ContentType（内容类型）
2. ContentEncoding（内容编码）
3. RemoteKey（路由键）
4. DeliveryMode（ 投递模式，消息是否持久化）
5. 等等...

有些属性只是约定规范，如ContentType，ContentEncoding，需要程序自己做处理，有些属性rabbitmq会根据值来进行处理，如RemoteKey，交换机会根据消息的RemoteKey和自身的类型来决定投递到哪些队列，DeliveryMode可以决定是否持久化消息。

#消息投递
ch.Publish(
    "",     # exchange名字，空为默认交换机
    key,    # routingkey 路由键
    false,  
    false,
    # 消息
    amqp.Publishing{
        DeliveryMode:amqp.Persistent,
        ContentType:"text/plain",
        Body:[]byte("hello world"),
    }) 

队列

队列是存储消息的主体，队列本身所拥有的一些属性：

1. Name 队列名字，不同的队列名字应该保持唯一性
2. Durable rabbitmq重启后，队列是否依旧存在，需要注意消息要持久化的要另外设置消息
3. exclusive 当前队列只能被一个消费者连接使用，关闭连接后删除队列。
4. auto-delete 最后一个消费者退订后删除队列。

在代码里面队列声明，如果队列不存在则新建队列，如果已存在相同名字的队列且属性不同的话则会报错。可以选择让系统自动生成队列，然后返回队列名字。

# 队列声明，参数依次为name，Durable，auto-delete，exclusive，no-wait.args
amqp.QueueDeclare("queuename", true, false, flase, false, nil)

消费者（consumer）

消费者用以消费队列里的消息的自定义程序片段，消费者获取队列里的消息有两种方式，一种是拉取（pull）的方式，通过channel.basicget方法，一种是订阅方式，队列推送消息到rabbitmq，这种方式用的最多。

消息处理

消息处理，消费者端连接队列后，可以得到一个类似于句柄的东西，如果没有消息就会一直阻塞。
消费者在收到消息之后处理的情况可能是成功的，也有可能是失败的，为了确保消息已经成功处理然后队列删除消息，如果失败则进行其他机制，以免消息一直重复在队列里面，或消息因消费者宕机而丢失。

消息确认（ack）

如果消息成功地被消费者处理的话，需要有一个消息确认的机制。
rabbitmq提供两种确认机制：

1. 自动确认模式，队列将消息发送给消费者之后立即删除消息（basic.deliver或basic.get-ok）
2. 显式确认模式，待消费者发送接受成功之后删除（basic.ack）

一般而言我们用的更多的是显式确认模式，如果消费者接收到消息没有进行确认之后就宕机了，队列里面的该消息还是会存在的，然后会把消息转发到其他消费者。

消息拒绝（basic.reject）

如果消费者对消息的处理出现了一些问题，可以调用rabbitmq的basic.reject来拒绝消息，拒绝消息之后，可以做的是把消息放回到队列里面，或者直接删除消息。

其实如果出现问题的消息，即便是交给其他的消费者，很会很大概率继续出现问题，这时候我们可以把消息放到其他专门处理记录问题的队列里面，交由另外的消费者处理。

交换机

交换机更像是消息的路由层，队列绑定到交换机，然后发布者可以发送的消息都是经过交换机的，然后经由消息的remote key（路由键）路由到交换机所绑定的队列里。
交换机分为4种类型：

1.直连交换机（direct）

直连型交换机（direct exchange）是根据消息携带的路由键（routing key）将消息投递给对应队列的。

1. 将一个队列绑定到某个交换机上，同时赋予该绑定一个路由键（routing key）
2. 当一个携带着路由键为R的消息被发送给直连交换机时，交换机会把它路由给绑定值同样为R的队列。

其实初尝rabbitmq的例子里面，看上去没有绑定交换机，实际上也是绑定了直连交换机，只是是一个特殊的预先声明好的，名字为空字符串的交换机，叫默认交换机，每个队列都会自动绑定到默认交换机上。

2.扇形交换机（funout）

扇型交换机（funout exchange）将消息路由给绑定到它身上的所有队列，而不理会绑定的路由键。如果N个队列绑定到某个扇型交换机上，当有消息发送给此扇型交换机时，交换机会将消息的拷贝分别发送给这所有的N个队列。扇型用来交换机处理消息的广播路由（broadcast routing）。

3.主题交换机（topic）

主题交换机（topic exchanges）通过对消息的路由键和队列到交换机的绑定模式之间的匹配，将消息路由给一个或多个队列。主题交换机经常用来实现各种分发/订阅模式及其变种。主题交换机通常用来实现消息的多播路由（multicast routing）。
主题交换机在我看来就像添加了简单的通配符+字符串来达到一个路由的规则。

4.头交换机（headers）

头交换机用的不是很多，有时消息的路由操作会涉及到多个属性，此时使用消息头就比用路由键更容易表达，头交换机（headers exchange）就是为此而生的。头交换机使用多个消息属性来代替路由键建立路由规则。通过判断消息头的值能否与指定的绑定相匹配来确立路由规则。

扇形路由器实现广播

生产者代码

conn, _ := amqp.Dial("amqp://localhost")

ch, _ := conn.Channel()

ch.ExchangeDeclare(
    "hello",
    "fanout",
    true,
    false,
    false,
    false,
    nil,
    )


ch.Publish(
    "hello",
    "",     // 由于是广播，所以可以不填写路由键
    false,
    false,
    amqp.Publishing{
        DeliveryMode:amqp.Persistent,
        Body:[]byte("hello"+time.Now().String()),
    })

消费者代码

conn, _ := amqp.Dial("amqp://localhost")

ch, _ := conn.Channel()
    
ch.ExchangeDeclare(
        "hello",  // 交换机名字
        "fanout", // 交换机类型
        true,     // durable
        false,    // autoDelete
        false,    // internal
        false,    // noWait
        nil,      // args
    )

q, _ := ch.QueueDeclare(
        "", 
        false,  // durable
        false,  // autoDelete
        true,   // exclusive
        false,  // noWait
        nil,    //
        )

ch.QueueBind(
    q.Name,     // queuename 
    "",         // remote key，由于是广播，可以不填写路由键
    "hello",    // exchange name
    false,      // nowait
    nil,
    )

msgs, _ := ch.Consume(q.Name,"", true, false, false,false,nil)

for msg := range msgs {
    log.Printf("%s", msg.Body)
}

主题交换机实现路由匹配

设有如下场景：设计一个日志收集系统，日志有不同的级别，debug，info，warn，error，日志格式为：

级别.模块名字 如：info.login

有不同的队列负责收集不同级别的日志，其中有个队列专门收集收集warn和error的数据，设计如下：
生产者

func main() {
    conn, _ := amqp.Dial("amqp://localhost")

    ch, _ := conn.Channel()


    ch.ExchangeDeclare(
        "logs",
        "topic",
        true,
        false,
        false,
        false,
        nil,
    )

    ch.Publish(
        "logs",
        "debug.123",
        false,
        false,
        amqp.Publishing{
            DeliveryMode:amqp.Persistent,
            Body:[]byte("hello"),
        },
    )
}

消费者

func main() {
    conn, _ := amqp.Dial("amqp://localhost")
    ch, _ := conn.Channel()

    ch.ExchangeDeclare(
        "logs",
        "topic",
        true,
        false,
        false,
        false,
        nil,
    )

    q, _ := ch.QueueDeclare(
        "log1",
        true,
        false,
        false,
        false,
        nil,
    )
    
    // 队列绑定的remote key
    keys := []string{"error.*", "warn.*"}

    for _, key := range keys{
        ch.QueueBind(
            q.Name,
            key,
            "logs",
            false,
            nil,
        )
    }

    deliver, _ := ch.Consume(
        q.Name,
        "",
        false,
        false,
        false,
        false,
        nil,
    )

    for d:= range deliver {
        fmt.Println(string(d.Body))
        d.Ack(true)
    }
}

rabbit入门教程

标签：情况   local   channel   rpc   图片   规则   rect   code   product   

原文地址：https://www.cnblogs.com/s-b-b/p/9069505.html