rabbitmq redis

RabbitMQ

RabbitMQ是一个在AMQP基础上完整的,可复用的企业消息系统。

对于RabbitMQ来说,生产和消费不再针对内存里的一个Queue对象,而是某台服务器上的RabbitMQ Server实现的消息队列

用rabbitmq实现一个简单的生产者消费者模型

发送端代码

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.1.25'))
channel = connection.channel()

channel.queue_declare(queue="hello")

channel.basic_publish(exchange='',
                     routing_key = 'hello',
                     body='hello world',
)
print("Send hello world")
connection.close()

接收端代码

 1 import pika
 2 
 3 connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.1.25'))
 4 channel = connection.channel()
 5 channel.queue_declare(queue="hello")
 6 
 7 def callback(ch,method,properties,body):
 8     print(ch,method,properties)
 9     print("received %s" %body)
10 
11 channel.basic_consume(callback,
12                       queue='hello',
13                       no_ack=True)
14 
15 print("waiting for messages to exit press 'CTRL+C'")
16 channel.start_consuming()

通过上述代码便可以实现一个简单的生产者消费者模型,但是现在的结果是:当开启多个消费者程序的时候,启动生产者发送消息,这个时候只有一个可以收到,并且再次启动,会下一个消费者收到,类似一个轮询的关系。

acknowledgment 消息不丢失(通过客户端设置实现)

通过no_ack = False参数设置,如果消费者遇到情况突然中断了没有收到,那么RabbitMQ会重新将任务添加到队列中

下面将接收端的代码进行更改:

#AUTHOR:FAN
import pika
import time

connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.1.25'))
channel = connection.channel()
channel.queue_declare(queue="hello")

def callback(ch,method,properties,body):
    print(ch,method,properties)
    time.sleep(10)
    print("received %s" %body)

channel.basic_consume(callback,
                      queue='hello',
                      no_ack=False)

print("waiting for messages to exit press 'CTRL+C'")
channel.start_consuming()

标注的地方就是代码修改的地方,通过将no_ack更改为False,以及在callback回到函数这里让等待10s,这样启动接收端后,再启动发送算,在还没有打印数据的时候将客户端关闭,然后再启动,发现依然可以收到刚才发送端发送的数据。

但是这种方式只能实现客户端断开重新连接的时候数据不丢失,如果是rabbitmq挂了的情况如何解决?

durable消息不丢失(通过在服务端设置保证数据不丢失)

这个时候生产者和消费者的代码都需要改动

发送者代码

 1 import pika
 2 
 3 
 4 connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.1.25'))
 5 channel = connection.channel()
 6 
 7 channel.queue_declare(queue='fan',durable=True)
 8 
 9 channel.basic_publish(exchange='',
10                       routing_key='fan',
11                       body='hello world',
12                       properties = pika.BasicProperties(
13                           delivery_mode=2
14                       ))
15 
16 print("send 'hello world'")
17 connection.close()

接收者的代码

 1 import pika
 2 import time
 3 
 4 connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.1.25'))
 5 channel = connection.channel()
 6 
 7 channel.queue_declare(queue='fan',durable=True)
 8 
 9 def callback(ch,method,properies,body):
10     print("received %s" %body)
11     time.sleep(10)
12     print("is ok")
13     ch.basic_ack(delivery_tag=method.delivery_tag)
14 
15 channel.basic_consume(callback,
16                       queue='fan',
17                       no_ack=False)
18 
19 print("waitting for messages.To exit press CTRL+C")
20 channel.start_consuming()

这样即使在接收者接收数据过程中rabbitmq服务器出现问题了,在服务恢复之后,依然可以收到数据

发布订阅

发布订阅和简单的消息队列区别在于,发布订阅会将消息发送给所有的订阅者,而消息队列中的数据被消费一次便消失。所以,RabbitMQ实现发布和订阅时,会为每一个订阅者创建一个队列,而发布者发布消息时,会将消息放置在所有相关队列中。

通过exchange type = fanout参数实现

代码例子:

发布者:

 1 #AUTHOR:FAN
 2 
 3 import pika
 4 import sys
 5 
 6 connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.8.103'))
 7 channel = connection.channel()
 8 
 9 channel.exchange_declare(exchange="fan",
10                          type='fanout')
11 
12 message = ' '.join(sys.argv[1:]) or "info :hello world"
13 channel.basic_publish(exchange = 'fan',
14                       routing_key='',
15                       body=message)
16 
17 print("send %s" %message)
18 connection.close()

订阅者:

#AUTHOR:FAN

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.8.103'))

channel = connection.channel()

channel.exchange_declare(exchange="fan",
                         type='fanout')

#随机生成队列名字
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue

#将exchange和队列绑定
channel.queue_bind(exchange='fan',
                   queue=queue_name)

print("waiting for fan ,To exit press CTRL+C")
def callback(ch,method,proerties,body):
    print("---",body)

channel.basic_consume(callback,
                      queue=queue_name,
                      no_ack=True)

channel.start_consuming()

关键字发送

通过参数:exchange type = direct实现

之前事例,发送消息时明确指定某个队列并向其中发送消息,RabbitMQ还支持根据关键字发送,即:队列绑定关键字,发送者将数据根据关键字发送到消息exchange,exchange根据 关键字 判定应该将数据发送至指定队列。

代码例子如下:

消费者代码:

 1 #AUTHOR:FAN
 2 import pika
 3 import sys
 4 connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.1.25'))
 5 channel = connection.channel()
 6 channel.exchange_declare(exchange='direct_logs_1',
 7                          type='direct')
 8 result = channel.queue_declare(exclusive=True)
 9 queue_name = result.method.queue
10 
11 severities = sys.argv[1:]
12 if not severities:
13     sys.stderr.write("Usage: %s [info] [warning] [error]\n" % sys.argv[0])
14     exit(1)
15 print(severities)
16 for severity in severities:
17     print(severity)
18     channel.queue_bind(exchange='direct_logs_1',
19     queue=queue_name,
20     routing_key=severity)
21 print("waiting for logs,To exit press CTRL+C")
22 def callback(ch,method,properties,body):
23     print("%s:%s" %(method.routing_key,body))
24 
25 channel.basic_consume(callback,
26                       queue=queue_name,
27                       no_ack=True)
28 channel.start_consuming()

生产者代码

 1 import pika
 2 import sys
 3 
 4 connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.1.25'))
 5 channel = connection.channel()
 6 
 7 channel.exchange_declare(exchange='direct_logs_1',
 8                          type='direct')
 9 
10 print(sys.argv)
11 severity = sys.argv[1] if len(sys.argv) >1 else "error"
12 message = ' '.join(sys.argv[2:]) or 'hello world'
13 channel.basic_publish(exchange='direct_logs_1',
14                       routing_key = severity,
15                       body = message)
16 print("send %s:%s" %(severity,message))
17 connection.close()

模糊匹配

通过参数exchange type = topic实现

在topic类型下,可以让队列绑定几个模糊的关键字,之后发送者将数据发送到exchange,exchange将传入”路由值“和 ”关键字“进行匹配,匹配成功,则将数据发送到指定队列。

# 表示可以匹配 0 个 或 多个 单词

*  表示只能匹配 一个 单词

--------------------还没有整理完

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏张戈的专栏

Linux运维工程师:30道面试题整理

前段时间,我在准备面试的时搜到的一套 Linux 运维工程师面试题,感觉比较全面,一直保存在草稿,刚在整理后台时翻了出来,干脆就发出来好了,以备不时之需。 1....

6365
来自专栏犀利豆的技术空间

Redis 命令的执行过程

之前写了一系列文章,已经很深入的探讨了 Redis 的数据结构,数据库的实现,key的过期策略以及 Redis 是怎么处理事件的。所以距离 Redis 的单机实...

561
来自专栏小樱的经验随笔

【批处理学习笔记】第二十七课:视窗

    首先我们要DIY它的窗口。使用命令:color、mode、msg。     --1、设置窗口背景色和字体颜色by color。详细内容参见color /...

2548
来自专栏极乐技术社区

用 RxJS、RxWX 编写微信小程序

RxJS RxJS是微软推出的ReactiveX系列,符合纯函数特点的第三方开源库有非常著名underscore和lodash,以及更加强大的RxJS。它可以用...

2948
来自专栏潇涧技术专栏

Head First Systrace

深入浅出systrace(1)systrace的简单介绍和systrace工具源码分析。

651
来自专栏JackieZheng

在ssh中利用Solr服务建立的界面化站内搜索---solr2

     继上次匆匆搭建起结合solr和nutch的所谓站内搜索引擎之后,虽当时心中兴奋不已,可是看了看百度,再只能看看我的控制台的打印出每个索引项的几行文字,...

1819
来自专栏程序员互动联盟

linux设备驱动第四篇:linux驱动调试方法

上一篇我们大概聊了如何写一个简单的字符设备驱动,我们不是神,写代码肯定会出现问题,我们需要在编写代码的过程中不断调试。在普通的c应用程序中,我们经常使用prin...

38815
来自专栏Sorrower的专栏

内核必须懂(一): 用系统调用打印Hello, world!

915
来自专栏Pythonista

Python操作mysql之模块pymysql

pymsql是Python中操作MySQL的模块,其使用方法和MySQLdb几乎相同。但目前pymysql支持python3.x而后者不支持3.x版本。

761
来自专栏Golang语言社区

Git 项目推荐 | Go 语言读写 INI 文件工具包

原文 http://git.oschina.net/Unknown/ini 主题 Git Go语言 本包提供了 Go 语言中读写 INI 文件的功能。 功能特...

45212

扫码关注云+社区