缓存，队列（Redis，RabbitMQ）
1、Redis
一、简介
Redis 与其余 key - value 缓存产品有如下三个特色：java
- Redis支持数据的持久化，能够将内存中的数据保存在磁盘中，重启的时候能够再次加载进行使用。
- Redis不只仅支持简单的key-value类型的数据，同时还提供list，set，zset，hash等数据结构的存储。
- Redis支持数据的备份，即master-slave模式的数据备份。
二、优点
- 性能极高 – Redis能读的速度是110000次/s,写的速度是81000次/s 。
- 丰富的数据类型 – Redis支持二进制案例的 Strings, Lists, Hashes, Sets 及 Ordered Sets 数据类型操做。
- 原子 – Redis的全部操做都是原子性的，意思就是要么成功执行要么失败彻底不执行。单个操做是原子性的。多个操做也支持事务，即原子性，经过MULTI和EXEC指令包起来。
- 丰富的特性 – Redis还支持 publish/subscribe, 通知, key 过时等等特性。
三、Redis与其余key-value存储有什么不一样？
-
Redis有着更为复杂的数据结构而且提供对他们的原子性操做，这是一个不一样于其余数据库的进化路径。Redis的数据类型都是基于基本数据结构的同时对程序员透明，无需进行额外的抽象。python
-
Redis运行在内存中可是能够持久化到磁盘，因此在对不一样数据集进行高速读写时须要权衡内存，由于数据量不能大于硬件内存。在内存数据库方面的另外一个优势是，相比在磁盘上相同的复杂的数据结构，在内存中操做起来很是简单，这样Redis能够作不少内部复杂性很强的事情。同时，在磁盘格式方面他们是紧凑的以追加的方式产生的，由于他们并不须要进行随机访问。mysql
四、Redis 安装
下载地址：https://github.com/MSOpenTech/redis/releases。git
Redis 支持 32 位和 64 位。这个须要根据你系统平台的实际状况选择，这里咱们下载 Redis-x64-xxx.zip压缩包到 C 盘，解压后，将文件夹从新命名为 redis程序员
打开一个 cmd 窗口 使用cd命令切换目录到 C:\redis 运行 redis-server.exe redis.windows.conf 。github
若是想方便的话，能够把 redis 的路径加到系统的环境变量里，这样就免得再输路径了，后面的那个 redis.windows.conf 能够省略，若是省略，会启用默认的。输入以后，会显示以下界面：web
这时候另启一个cmd窗口，原来的不要关闭，否则就没法访问服务端了。redis
切换到redis目录下运行 redis-cli.exe -h 127.0.0.1 -p 6379 。sql
设置键值对 set myKey abcshell
取出键值对 get myKey
五、安装目录的文件
redis-server : 服务器
redis-cli ：命令行客户端
redis-benchmark 性能工具测试
redis-check-aof ADF文件修复工具
redis-check-dump RDB文件检测工具
redis.conf是redis的配置文件
将配置文件中的daemonize yes 以守护进程的方式来使用 cd到redis的安装目录下 启动和中止 启动 ： redis-server 中止：shutdown 命令返回值 1）状态回复 pong set test "this is a test" 2）错误回复 (error) ERR unknown command 'testerror' 3）整数回复： (integer) 4 4）字符串回复 get 'test' (nil) 表明空的结果 5）多行字符串回复 KEYS * 获得当前数据库中的存在的键值名
六、Redis配置选项相关的内容
1>动态设置/获取配置选项的值 获取 CONFIG GET port 1）"post" 2)"6379" 动态设置 CONFIG GET warning 2>Redis配置文件redis.conf选项相关 ---- 链接选项 ---- port 6379 默认端口 bind 127.0.01 默认绑定的主机地址 timeout 0 客户端闲置多久后自动关闭链接 0 表明没有启动这个选项 loglevel notice 日志的记录级别 debug 调试的很详细的信息：适合开发和测试 verbose 包含不少不太有用的信息 notice 生产环境 warning 警告信息 logfile stdout 指定日志的记录方式 默认是标准输出 database 16 设置默认数据库的数量，默认是16个 SETECT 1 选择数据库 默认编号是0 ----- 快照相关 ------ 多少秒有多少次改变将其同步到磁盘中的数据文件中 save 900 1 900表明秒数 900秒内有一个更改就记录到磁盘中 save 300 10 save 60 10000 rdbcompression yes 存储本地数据库是是否启动压缩， 默认yes dbfilename dump.db 指定本地数据库的文件名
七、Redis的数据类型
1》String类型
1 一个键最多存储512MB 2 1》SET 设置key 对应的值为value 3 4 语法：SET key value [EX seconds] [PX milliseconds] [NX|XX] 5 6 EX seconds 设置key的过时时间 SET key value EX seconds == SETEX 7 PX milliseconds 以毫秒的形式设置过时时间 SET key value PX milliseconds -- PSETEX 8 NX ：只有键不存在的时候才会设置成功 --SETNX 9 XX :只有key已经存在的时候才能够设置 10 11 SET test16 'this is a test' EX 100 12 SET test17 'this is a test17' PX 20000 13 SET test18 'this is a test18' NX 14 SET test18 'this is a test18888' XX 15 16 SET test19 'this is a test19' EX 100 NX 17 18 SET test20 'this is a test20' EX 100 PX 300000 NX 19 20 21 注意： 若是key存在，同名会产生覆盖 22 23 SET testStr1 "this is a test" 24 25 2》GET key 根据key找到对应的值 26 27 语法： GET key 28 29 注意：若是key不存在，返回nil 30 若是key不是字符串，会报错 31 32 3》GETGANGE: 返回字符串中的一部分 33 34 语法：GETRANGE key start end 35 36 GETGANGET testStr2 0 4 37 GETGANGET testStr2 0 -3 38 GETGANGET testStr2 -4 -2 39 GETGANGET testStr2 0 1000 40 41 4》GETSET:设定指定能够的值，返回原来的值 42 计数器的清零效果 43 44 语法：GETSET key value 45 SET testStr3 'king' 46 47 GET testStr3 48 49 GETSET testStr3 'queen' 50 51 注意：当key不存在返回nil 若是key不是字符串会报错 52 53 5》MSET 一次设置多个 54 55 语法：MSET key value [key value...] 56 MSET testStr5 'king' testStr6 'sk' testStr7 'queen' 57 58 6》MGET 一次得到多个 59 60 语法：MGET key key 61 MGET testStr5 testStr6 testStr7 62 63 7》STRLEN:获取key的字符串的长度 64 65 语法：STRLEN key 66 67 注意：对于不存在的key获取其长度返回的是0 68 69 8》SETRANGE:至关于字符串替换的效果 70 71 语法：8》SETRANGE key offset value 72 73 注意：若是设置的key原来的字符串长度要比偏移量小，就会以零字节（\x00）来填充 74 75 SET testStr9 'hello king' 76 77 SETRANGE testStr9 6 'queen' 78 79 对于不存在的key使用SETRANGE 80 81 EXISTS testStr10 检测key是否存在 82 83 SETRANGE testStr10 5 'king' 84 85 9》SETNX 只有key不存在才能设置成功 86 87 语法：SETNX key value 88 89 10》SETEX:key 设置key而且设置过时时间，以秒为单位 90 91 语法：SETEX key seconds value 原子性操做 92 TTL 获得键的生存周期 93 94 注意：SETEX 是原子性操做，至关于执行了SET key value ，又对这个key设置了过时时间 EXPIRE key seconds 95 96 97 SET expireStr1 'test1' 98 99 EXPIRE expireStr1 10 100 101 SETEX test12 1000 'a' 102 103 GET test12 104 105 11》 MSETNX 一次设置多个key-value ,只有全部的key都不存在的时候才会成功 106 107 语法 MSETNX key value [key value] 108 109 MSETNX test13 'a' test14 'b' test15 'c' 110 111 12》PSETEX:以毫秒为单位设置key的生存周期 112 113 语法：PSETEX key milliseconds value 114 115 PSETEX test16 2000 'hell0 king' 116 117 PTTL 118 13》INCR 对key中存储的数字+1 119 120 语法：INCR key 121 122 SET counter 1 123 124 INCR counter 125 126 注意：key若是不存在会先初始化为0，在进行INCR操做， 127 对于不是数值的值会报错 128 129 14》INCR BY ： 将key中存储的数字加上指定增量 130 131 语法：INCRBY key INCREMENT 132 133 SET counter2 10 134 INCRBY counter2 5 135 INCRBY counter2 1.2 不能用浮点数 136 15》INCRBYFLOAT : 给key中存储的数字加上指定的浮点数 137 138 语法：INCRBYFLOAT key increment 139 140 SET counter3 1 141 142 INCRBYFLOAT counter3 1.2 143 144 16》DECR:给将key中存储的数字减1 145 146 语法：DECR key 147 148 17》DECYBY:将key中存储的数值减去指定的值 149 150 语法：DECRBY key decrement 151 152 18》APPEND：将值追加到字符串的末尾 153 154 语法：APPEND key value
2》Hash类型
1 Hash类型（散列表） 2 在配置文件中能够经过配置 3 hash-max-ziplist-entries 512 512个字节 4 hash-max-ziplist-value 64 字段数目 5 6 Hash相关命令 7 1》HSET : 将哈希表key中域field设置成指定的value 8 9 语法 ： HSET key field value 10 11 HSET userInfo1 username 'king' 12 13 HSET userInfo1 password '123456' 14 15 HSET userInfo1 email '18368827317@163.com' 16 17 HSET username 'queen' 18 19 注意：若是哈希表中key中field不存在至关于新建field，设置成功返回1 20 若是哈希表中key中field存在，至关于从新赋值，成功返回0 21 22 2》HGET :返回哈希表中给定field的值 23 24 语法：HGET key field 25 26 HGET userInfo1 username 27 28 注意：若是key中field不存在，返回的是nil 29 30 3》HSETNX: 将哈希表中的field设定成指定的值，只有field不在的时候才能成功，若是filed存在，操做无效 31 32 语法：HSETNX key filed value 33 34 HSETNX testHash1 test 'a' 35 36 4》 HMSET :经过将多个field-value设置到hash表key中 37 38 语法：HMSET key value filed value ... 39 40 HMSET userInfo2 username（域） 'king'（值） kickname 'freeyman' 41 42 5》HMGET ： 一次得到hash表key中多个filed的值 43 44 语法：HMGET key field field 45 46 注意：若是hash表中field不存在，会返回nil 47 48 6》HGETALL:返回哈希表key中全部的field和value 49 50 语法：HGETALL key 51 52 7》HKEYS:返回hash中key的全部的filed 53 54 语法： HKEYS key 55 56 8》HVALS ：返回全部hash中key中field全部的对应的值 57 58 语法：HVALS key 59 60 9》HEXISTS :检测hash表中key的field是否存在 61 62 语法：HEXISTS key field 63 64 HEXISTS userInfo2 username 65 66 HEXISTS userInfo2 notExists 67 68 10》HLEN: 返回hash表中keyfield的数量 69 70 语法：HLEN key 71 72 11》HINCRBY:给hash中key的field作增量操做 73 74 语法：HINCRBY key field increment 75 76 12》HINCRBYFLOAT: 给hash中key的field作增量操做 家浮点数 77 78 语法：HINCRBYFLOAT key field increment 79 80 HSET userInfo3 salary '12343.341' 81 82 HINCRBYFLOAT userInfo3 salary '0.214' 83 84 13》HDEL ：删除hash中key的指定域，能够删除一个也能够删除多个 85 86 语法：HDEL key field field 87 88 HGETALL userInfo2 89 90 HDEL userInfo2 username
3》List类型
1 双向链表实现的两边的获取速度快） 2 3 1》LPUSH：向列表左端添加元素 4 5 语法：LPUSH key value... 从左到右依次插入 6 7 LPUSH myList1 a b c 8 9 2》RPUSH ：向列表右端添加元素 10 11 语法：RPUSH key value... 12 13 RPUSH myList1 test1 test2 test3 14 15 3》LPUSHX：向列表头部添加元素，只有key存在在来添加 只能添加一个值， 16 17 语法：LPUSHX key value 18 19 LPUSH myList2 test4 20 21 4》RPUSHX：向列表尾部添加元素，只有key存在在来添加 22 23 语法：RPUSHX key value 24 25 RPUSH myList2 test4 26 27 5》LPOP ：将列表头部的元素弹出 28 29 语法：LPOP myList1 30 31 6》RPOP : 将列表尾部的元素弹出 32 33 语法：RPOP myList1 34 35 7》LLEN : 获得列表的长度 36 37 语法：LLEN key 38 39 8》LRANGE : 获取列表片断 40 41 语法：LRANGE key start stop 42 43 LRANGE myList1 0 -1 44 45 注意：若是start下标比列表的最大的下标end大，返回空列表 46 若是stop比列表长度大，返回列表的末尾 47 48 9》LREM:删除列表中指定的值 49 50 语法：LRME key count value 51 52 count>0 从列表的头开始，向尾部搜索，移除与value相等的元素，移动count个 53 54 count<0 从列表的尾部向头搜索，移除与value相等的元素，移除count个 55 56 count=0 移除列表全部与count相等的值 57 58 LPUSH myList3 a b c d a b c d b e f b g e b 59 60 LREM myList3 2 b 61 62 LREM myList3 -1 a 63 64 LREM myList3 0 e 65 66 10》LINDEX:获取指定索引元素的值 67 68 语法：LINDEX key index 69 70 LINDEX myList3 3 71 72 LINDEX myList3 -2 73 74 11》LSET ：设置指定索引元素的值 75 76 LSET key index value 77 78 12》LTRIM :只保留列表的片断 79 80 语法：LTRIN key start stop 81 82 LPUSH myList4 log1 log2 log3 log4 log5 83 84 LRTIM myList4 0 1 85 86 LPUSH myList4 a b c d e f g 87 88 LTRIM myList4 1 -1 89 90 LTRIM myList4 1 91 92 13》LINSERT 向列表插入元素 93 94 语法：LINSERT key BEFORE|AFTER pivot value 95 96 LPUSH myList6 a b c d 97 98 LINSERT myList6 BEFORE "b" "king" 99 100 若是没有成功返回-1 成功返回当前列表的长度 对于空列表返回0 不成功 101 102 14》RPOPLPUSH:将一个元素从一个列表转移到另外一个列表（原子性操做） 103 104 语法：RPOPLPUSH source destination 105 106 RPOPLPUSH myList1 myList6 107 108 15》BLPOP:BLPOP是LPOP 的一个阻塞版本 109 110 语法： BLPOP key [key...] timeout 111 112 LPUSH myList9 a b c 113 114 LPUSH myList10 d e f 115 116 BLPOP myList8 myList9 myList10 0 117 118 BLPOP myList8 8 0
4》Set类型
1 sns 和 博客系统 能够经过集合类型实现 2 3 1》SADD ：向集合中添加元素 4 5 语法：SADD key member [,...] 6 7 SADD web sunkai.clog.com 8 9 2》SMEMBERS :返回指定集合中的元素 10 11 语法：SMEMBERS key 12 13 3》SISMEMBER ： 检测member是不是集合中的成员 14 15 语法：SISMEMBER key member 16 17 4》SREM ：删除 18 19 语法：SREM key member 20 21 5》SPOP ：随机删除并返回集合中的删除的元素 22 23 语法：SPOP key 24 25 6》SRANDMEMBER :随机返回集合中的元素 26 27 语法：SRANDMEMBER key counter 28 29 注意：count 为正数，并且小于集合中的元素，返回的一个包含随机元素的集合数组；count若是大于集合中的元素，这个时候会返回整个集合 30 count 为负数，返回一个数组，数组中的成员可能出现重复，数组的长度是count取绝对值 31 32 7》SDIFF 返回集合间的差集 33 34 语法：SDIFF key key 35 36 SADD couser2 java PHP js jq Python 37 38 SADD couser1 iOS anzhuo Python 39 40 SDIFF couser2 couser1 41 42 8》SINTER 返回集合间的交集 43 44 语法：SINTER key key ... 45 46 SINTER couser2 couser1 47 48 9》SUNION ：返回集合间的并集 49 50 语法：SUNION key key 51 52 10》SCARD ：返回集合的长度 53 54 语法：SCARD 55 56 11》SDIFFSTORE ：讲差集结果保存到指定集合中 57 58 语法：SDIFFSTORE destination key key ... 59 60 SDIFFSTORE diffSET couser2 couser1 61 62 12》SINTERSTORE 63 64 13》SUNIONSTORE 65 66 14》SMOVE 将集合中的元素移动到另外一个集合中（原子性操做） 67 68 语法：SMOVE source destination member
5》Zset（sorted set）有序集合类型
1 1》ZADD :将元素及其分数添加到集合中 2 3 语法： ZADD key score member [score member] 4 5 ZADD PYTHONcourse 100 king 6 7 ZADD PYTHONcourse 98 queen 98 test 78 test1 8 9 ZADD PYTHONcourse +inf maxInt -inf minInx 正无穷大，负无穷大 10 11 2》ZSCORE ：得到集合中的指定元素的分数 12 13 语法：ZSCORE key member 14 15 3》ZRANGE ：按照元素分数从小到大的顺序返回指定索引start到stop之间的全部元素（包含两端） 16 17 语法：ZRANGE key start stop WITHSCORES 带分数 18 19 注意：当元素的两个元素的分数相同的时候，redis在排序按照字典的顺序排列 20 21 4》ZREVRANGE: 和ZRANGE 相反，按照从大到小的顺序返回 22 23 语法：ZREVRANGE key start stop [WITHSCORES] 24 25 5》ZRANGEBYSCORE ：获取指定分数范围内的元素，按照从小到大的顺序，返回的是分数在指定的min到max之间 26 27 语法：ZRANGEBYSCORE key min max [WITHSCORES] [LIMIT offset count] 28 29 得到分数80~90之间的全部元素 30 31 ZRANGEBYSCORE PYTHONcourse 80 90 32 33 ZRANGEBYSCORE PYTHONcourse 80 (90 不包含90 34 35 注意：经过左括号表明不包含端点 36 37 6：ZREVRANGEBYSCORE 和上面的相反 38 39 语法：ZREVRANGEBYSCORE key min max [WITHSCORES] [LIMIT offset count] 40 41 7》ZINCRBY : 操做某个元素的分数，返回操做以后的分数 42 43 语法：ZINCRBY key increment member 44 45 8》ZCARD :得到集合中元素的数量 46 47 语法：ZCARD key 48 49 9》ZCOUNT 得到指定分数内元素的数量 50 51 语法：ZCOUNT PYTHONcourse 80 90 52 53 10》ZREM ：删除一个或多个元素，返回删除元素的个数 54 55 语法：ZREM key member ... 56 57 11》ZREMRANGEBYRANK: 按照排名范围删除元素，按照分数从小到大的顺序删除全部指定的排名范围内的全部元素 58 59 语法：ZREMRANGEBYRANK key start stop 60 61 12》ZREMRANGEBYSCORE:按照分数范围内删除元素 62 63 语法：ZREMRANGEBYSCORE key min max 64 65 13》ZRANK :得到指定元素的排名，根据分数从小到大的顺序 66 67 语法：ZRANK key member 68 69 14》ZREVRANK:得到指定元素的排名，根据分数从大到小的顺序 70 71 语法：ZREVRANK key member 72 73 15》ZINTERSTORE: 计算有序集合的交集，并将结果保存起来 74 75 语法：ZINTERSTORE destination numkeys key key ... WEIGHTS weight weight AGGREGATE [SUM|MIN|MAX] 76 77 ZADD testSortedSet1 1 a 2 b 3 c 78 79 ZADD testSortedSet2 10 a 20 b 30 c 80 81 ZINTERSTORE testSorted1 2 testSortedSet1 testSortedSet2 82 83 ZRANGE testSorted1 0 -1 WITHSCORES 84 85 ZINTERSTORE testSorted2 2 testSortedSet1 testSortedSet2 AGGREGATE SUM 86 87 ZINTERSTORE testSorted3 2 testSortedSet1 testSortedSet2 AGGREGATE MIN 88 89 16》ZUNIONSTORE :计算有序集合并集，并将结果保存起来 90 91 语法：ZUNIONSTORE destination numkeys key key ... WEIGHTS weight weight AGGREGATE [SUM|MIN|MAX]
未完待续。。。。。。
2、RabbitMQ
一、简介
RabbitMQ是一个在AMQP基础上完整的，可复用的企业消息系统。他遵循Mozilla Public License开源协议。
MQ全称为Message Queue, 消息队列（MQ）是一种应用程序对应用程序的通讯方法。应用程序经过读写出入队列的消息（针对应用程序的数据）来通讯，而无需专用链接来连接它们。消息传递指的是程序之间经过在消息中发送数据进行通讯，而不是经过直接调用彼此来通讯，直接调用一般是用于诸如远程过程调用的技术。排队指的是应用程序经过队列来通讯。队列的使用除去了接收和发送应用程序同时执行的要求。RabbitMQ能够，多个程序同时使用RabbitMQ ,可是必须队列名称不同。采用erlang语言，属于爱立信公司开发的。
消息中间件 --->就是消息队列
异步方式：不须要立马获得结果，须要排队
同步方式：须要实时得到数据，坚定不能排队
subprocess 的Q也提供不一样进程之间的沟通
应用场景：
电商秒杀活动
抢购小米手机
堡垒机批量发送文件
二、Centos6.x系统编译安装RabbitMQ
1、安装erlang
依赖包：
yum -y install gcc ncurses ncurses-base ncurses-devel ncurses-libs ncurses-static ncurses-term ocaml-curses ocaml-curses-devel openssl-devel zlib-devel openssl-devel perl xz xmlto kernel-devel m4 这是一行
一、下载otp_src_19.3.tar.gz
二、tar xvf otp_src_19.3.tar.gz
三、cd opt_src_19.3.tar.gz
四、./configure --prefix=/usr/local/erlang --with-ssl --enable-threads --enable-smp-support --enable-kernel-poll --enable-hipe --without-javac
五、make && make install
五、配置erlang环境：
vi /etc/profile
export PATH=$PATH:/usr/local/erlang/bin
source /etc/profile # 环境变量重启生效
2、安装rabbitmq
一、下载rabbitmq-server-generic-unix-3.6.5.tar.xz
二、tar xvf rabbitmq-server-generic-unix-3.6.5.tar.xz
三、mv rabbitmq_server-3.6.5/ /usr/local/rabbitmq
四、启动：
#启动rabbitmq服务
/usr/local/rabbitmq/sbin/rabbitmq-server
#后台启动
/usr/local/rabbitmq/sbin/rabbitmq-server -detached
#关闭rabbitmq服务
/usr/local/rabbitmq/sbin/rabbitmqctl stop
或
ps -ef | grep rabbit 和 kill -9 xxx 杀死服务
#开启插件管理页面
/usr/local/rabbitmq/sbin/rabbitmq-plugins enable rabbitmq_management
#建立用户
/usr/local/rabbitmq/sbin/rabbitmqctl add_user rabbitadmin 123456
/usr/local/rabbitmq/sbin/rabbitmqctl set_user_tags rabbitadmin administrator
./rabbitmqctl set_permissions -p / rabbitadmin ".*" ".*" ".*" 为这个用户受权
五、登陆
#WEB登陆
http://10.10.3.63:15672 本身的IP地址
用户名：rabbitadmin
密码：123456
3、几种队列通讯
一、实现最简单的队列通讯
sender
import pika
# 认证
credentials = pika.PlainCredentials('rabbitadmin', '123456') # 必定要认证
# 链接这台机器
connection = pika.BlockingConnection(pika.ConnectionParameters(
'192.168.14.38',credentials=credentials)) #主机IP 和验证
channel = connection.channel() # 创建了rabbitmq的协议通道
# 声明queue队列
channel.queue_declare(queue='hello')
# n RabbitMQ a message can never be sent directly to the queue, it always needs to go through an exchange.
# 发送消息
channel.basic_publish(exchange='', exchange表示交换器，能精确指定消息应该发送到哪一个队列，
routing_key='hello', #
body='Hello World!')#routing_key设置为队列的名称，body就是发送的内容
print(" [x] Sent 'Hello World!'")
connection.close()
import pika
import time
credentials = pika.PlainCredentials('rabbitadmin', '123456')
# 链接这台机器
connection = pika.BlockingConnection(pika.ConnectionParameters(
'192.168.14.38',credentials=credentials))
channel = connection.channel() # 创建了rabbitmq的协议通道
channel.queue_declare(queue='hello')
def callback(ch, method, properties, body):
print("received msg...start processing....",body)
time.sleep(20)
print(" [x] msg process done....",body)
channel.basic_consume(callback,
queue='hello',
no_ack=True)
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()
二、Work Queues （一个发消息，两个收消息，收消息是公平的依次分发）
在这种模式下，RabbitMQ会默认把p发的消息依次分发给各个消费者(c),跟负载均衡差很少。
消息提供者代码
import pika
import time
connection = pika.BlockingConnection(pika.ConnectionParameters(
'localhost'))
channel = connection.channel()
# 声明queue
channel.queue_declare(queue='task_queue')
# n RabbitMQ a message can never be sent directly to the queue, it always needs to go through an exchange.
import sys
message = ' '.join(sys.argv[1:]) or "Hello World! %s" % time.time()
channel.basic_publish(exchange='',
routing_key='task_queue',
body=message,
properties=pika.BasicProperties(
delivery_mode=2, # make message persistent
)
)
print(" [x] Sent %r" % message)
connection.close()
import pika, time
connection = pika.BlockingConnection(pika.ConnectionParameters(
'localhost'))
channel = connection.channel()
def callback(ch, method, properties, body):
print(" [x] Received %r" % body)
time.sleep(20)
print(" [x] Done")
print("method.delivery_tag",method.delivery_tag)
ch.basic_ack(delivery_tag=method.delivery_tag)
channel.basic_consume(callback,
queue='task_queue',
no_ack=True
)
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()
此时，先启动消息生产者，而后再分别启动3个消费者，经过生产者多发送几条消息，你会发现，这几条消息会被依次分配到各个消费者身上
三、消息持久化
虽然有了消息反馈机制，可是若是rabbitmq自身挂掉的话，那么任务仍是会丢失。因此须要将任务持久化存储起来。声明持久化存储：
将队列（Queue）与消息（Message）都设置为可持久化的（durable），这样能够保证绝大部分状况下咱们的RabbitMQ消息不会丢失。但依然解决不了小几率丢失事件的发生（好比RabbitMQ服务器已经接收到生产者的消息，但还没来得及持久化该消息时RabbitMQ服务器就断电了），若是须要对这种小几率事件也要管理起来，那么要用到事务。因为这里仅为RabbitMQ的简单介绍，因此不讲解RabbitMQ相关的事务。
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel() # 创建了rabbit 协议的通道
# durable=True 声明持久化存储
channel.queue_declare(queue='task_queue', durable=True)
channel.basic_publish(exchange='',
routing_key='task_queue',
body='Hello World!',
# 在发送任务的时候，用delivery_mode=2来标记消息为持久化存储
properties=pika.BasicProperties(
delivery_mode=2,
))
print(" [x] Sent 'Hello World!'")
connection.close()
sender.py
import pika
import time
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='task_queue', durable=True)
def callback(ch, method, properties, body):
print("received msg...start processing....",body)
time.sleep(20)
print(" [x] msg process done....", body)
ch.basic_ack(delivery_tag=method.delivery_tag)
channel.basic_consume(
callback,
queue='task_queue',
no_ack=False # 默认为False
)
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()
四、消息公平分发
若是Rabbit只管按顺序把消息发到各个消费者身上，不考虑消费者负载的话，极可能出现，一个机器配置不高的消费者那里堆积了不少消息处理不完，同时配置高的消费者却一直很轻松。为解决此问题，能够在各个消费者端，配置perfetch=1,意思就是告诉RabbitMQ在我这个消费者当前消息还没处理完的时候就不要再给我发新消息了。
channel.basic_qos(prefetch_count=1)
带消息持久化+公平分发的完整代码
生产者端
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.queue_declare(queue='task_queue', durable=True)
message = ' '.join(sys.argv[1:]) or "Hello World!"
channel.basic_publish(exchange='',
routing_key='task_queue',
body=message,
properties=pika.BasicProperties(
delivery_mode = 2, # make message persistent
))
print(" [x] Sent %r" % message)
connection.close()
消费者端
import pika
import time
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.queue_declare(queue='task_queue', durable=True)
print(' [*] Waiting for messages. To exit press CTRL+C')
def callback(ch, method, properties, body):
print(" [x] Received %r" % body)
time.sleep(body.count(b'.'))
print(" [x] Done")
ch.basic_ack(delivery_tag = method.delivery_tag)
channel.basic_qos(prefetch_count=1)
channel.basic_consume(callback,
queue='task_queue')
channel.start_consuming()
五、Publish\Subscribe(消息发布\订阅)
以前的例子都基本都是1对1的消息发送和接收，即消息只能发送到指定的queue里，但有些时候你想让你的消息被全部的Queue收到，相似广播的效果，这时候就要用到exchange了，
交流是一个很是简单的事情。一方面它收到消息从生产者和另外一边推他们队列。交换必须知道如何处理接收到的消息。应该是附加到一个特定的队列吗?应该是附加到多队列?或者应该丢弃。交换的规则定义的类型。
Exchange在定义的时候是有类型的，以决定究竟是哪些Queue符合条件，能够接收消息
fanout: 全部bind到此exchange的queue均可以接收消息
direct: 经过routingKey和exchange决定的那个惟一的queue能够接收消息
topic:全部符合routingKey(此时能够是一个表达式)的routingKey所bind的queue能够接收消息
表达式符号说明：#表明一个或多个字符，*表明任何字符
例：#.a会匹配a.a，aa.a，aaa.a等
*.a会匹配a.a，b.a，c.a等
注：使用RoutingKey为#，Exchange Type为topic的时候至关于使用fanout
headers: 经过headers 来决定把消息发给哪些queue
消息publisher
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='logs',
type='fanout')
message = ' '.join(sys.argv[1:]) or "info: Hello World!"
channel.basic_publish(exchange='logs',
routing_key='',
body=message)
print(" [x] Sent %r" % message)
connection.close()
消息subscriber
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='logs',
type='fanout')
result = channel.queue_declare(exclusive=True) #不指定queue名字,rabbit会随机分配一个名字,exclusive=True会在使用此queue的消费者断开后,自动将queue删除
queue_name = result.method.queue
channel.queue_bind(exchange='logs',
queue=queue_name)
print(' [*] Waiting for logs. To exit press CTRL+C')
def callback(ch, method, properties, body):
print(" [x] %r" % body)
channel.basic_consume(callback,
queue=queue_name,
no_ack=True)
channel.start_consuming()
六、有选择的接收消息(exchange type=direct)
RabbitMQ还支持根据关键字发送，即：队列绑定关键字，发送者将数据根据关键字发送到消息exchange，exchange根据 关键字 断定应该将数据发送至指定队列
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='direct_logs',
type='direct')
severity = sys.argv[1] if len(sys.argv) > 1 else 'info'
message = ' '.join(sys.argv[2:]) or 'Hello World!'
channel.basic_publish(exchange='direct_logs',
routing_key=severity,
body=message)
print(" [x] Sent %r:%r" % (severity, message))
connection.close()
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='direct_logs',
type='direct')
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
severities = sys.argv[1:]
if not severities:
sys.stderr.write("Usage: %s [info] [warning] [error]\n" % sys.argv[0])
sys.exit(1)
for severity in severities:
channel.queue_bind(exchange='direct_logs',
queue=queue_name,
routing_key=severity)
print(' [*] Waiting for logs. To exit press CTRL+C')
def callback(ch, method, properties, body):
print(" [x] %r:%r" % (method.routing_key, body))
channel.basic_consume(callback,
queue=queue_name,
no_ack=True)
channel.start_consuming()
七、更细致的消息过滤
Although using the direct exchange improved our system, it still has limitations - it can't do routing based on multiple criteria.
In our logging system we might want to subscribe to not only logs based on severity, but also based on the source which emitted the log. You might know this concept from the syslog unix tool, which routes logs based on both severity (info/warn/crit...) and facility (auth/cron/kern...).
That would give us a lot of flexibility - we may want to listen to just critical errors coming from 'cron' but also all logs from 'kern'.
topi: 意思是话题
To receive all the logs run:
python receive_logs_topic.py "#" #绑定#号，就是收全部消息，至关于广播
To receive all logs from the facility "kern":
python receive_logs_topic.py "kern.*" #以kern开头
Or if you want to hear only about "critical" logs:
python receive_logs_topic.py "*.critical" #以critical结尾
You can create multiple bindings:
python receive_logs_topic.py "kern.*" "*.critical" #收kern开头而且以critical结尾（至关于收两个）
And to emit a log with a routing key "kern.critical" type:
python emit_log_topic.py "kern.critical" "A critical kernel error" #发消息到kern.critical里,内容是：
A critical kernel error
示例：
rabbit_topic_send.py （生产者是发送端）
1 import pika
2 import sys
3
4 credentials = pika.PlainCredentials('nulige', '123456')
5 connection = pika.BlockingConnection(pika.ConnectionParameters(
6 host='192.168.1.118',credentials=credentials))
7
8 channel = connection.channel()
9
10 channel.exchange_declare(exchange='topic_logs',type='topic') #指定类型
11
12 routing_key = sys.argv[1] if len(sys.argv) > 1 else 'anonymous.info'
13
14 message = ' '.join(sys.argv[2:]) or 'Hello World!' #消息
15
16 channel.basic_publish(exchange='topic_logs',
17 routing_key=routing_key,
18 body=message)
19 print(" [x] Sent %r:%r" % (routing_key, message))
20 connection.close()
rabbit_topic_recv.py （消费者是接收端）单向的
1 import pika
2 import sys
3
4 credentials = pika.PlainCredentials('nulige', '123456')
5 connection = pika.BlockingConnection(pika.ConnectionParameters(
6 host='192.168.1.118',credentials=credentials))
7
8 channel = connection.channel()
9 channel.exchange_declare(exchange='topic_logs',type='topic')
10
11 result = channel.queue_declare(exclusive=True)
12 queue_name = result.method.queue
13
14 binding_keys = sys.argv[1:]
15 if not binding_keys:
16 sys.stderr.write("Usage: %s [binding_key]...\n" % sys.argv[0])
17 sys.exit(1)
18
19 for binding_key in binding_keys:
20 channel.queue_bind(exchange='topic_logs',
21 queue=queue_name,
22 routing_key=binding_key)
23
24 print(' [*] Waiting for logs. To exit press CTRL+C')
25
26 def callback(ch, method, properties, body):
27 print(" [x] %r:%r" % (method.routing_key, body))
28
29 channel.basic_consume(callback,queue=queue_name)
30
31 channel.start_consuming()
执行结果：
1 #接收端 2 D:\python\day42>python3 rabbit_topic_recv.py error 3 [*] Waiting for logs. To exit press CTRL+C 4 [x] 'error':b'mysql has error' 5 6 7 D:\python\day42>python3 rabbit_topic_recv.py *.warning mysql.* 8 [*] Waiting for logs. To exit press CTRL+C 9 [x] 'mysql.error':b'mysql has error' 10 11 12 D:\python\day42>python3 rabbit_topic_send.py mysql.info "mysql has error" 13 [x] Sent 'mysql.info':'mysql has error' 14 15 16 D:\python\day42>python3 rabbit_topic_recv.py *.error.* 17 [*] Waiting for logs. To exit press CTRL+C 18 [x] 'mysql.error.':b'mysql has error' 19 20 21 #发送端 指定类型:error 消息内容 22 D:\python\day42>python3 rabbit_topic_send.py error "mysql has error" 23 [x] Sent 'error':'mysql has error' 24 25 26 D:\python\day42>python3 rabbit_topic_send.py mysql.error "mysql has error" 27 [x] Sent 'mysql.error':'mysql has error' 28 [x] 'mysql.info':b'mysql has error' 29 30 31 D:\python\day42>python3 rabbit_topic_send.py mysql.error. "mysql has error" 32 [x] Sent 'mysql.error.':'mysql has error'
八、Remote procedure call (RPC) 双向的
To illustrate how an RPC service could be used we're going to create a simple client class. It's going to expose a method named call which sends an RPC request and blocks until the answer is received:
fibonacci_rpc = FibonacciRpcClient()
result = fibonacci_rpc.call(4)
print("fib(4) is %r" % result)
应用场景：
示例：实现RPC服务功能
rabbit_rpc_send.py（生产者是发送端）
1 import pika
2 import uuid
3
4 class SSHRpcClient(object):
5 def __init__(self):
6 credentials = pika.PlainCredentials('nulige', '123456')
7 self.connection = pika.BlockingConnection(pika.ConnectionParameters(
8 host='192.168.1.118',credentials=credentials))
9
10 self.channel = self.connection.channel()
11
12 result = self.channel.queue_declare(exclusive=True) #客户端的结果必需要返回到这个queue
13 self.callback_queue = result.method.queue
14
15 self.channel.basic_consume(self.on_response,queue=self.callback_queue) #声明从这个queue里收结果
16
17 def on_response(self, ch, method, props, body):
18 if self.corr_id == props.correlation_id: #任务标识符
19 self.response = body
20 print(body)
21
22 # 返回的结果，放在callback_queue中
23 def call(self, n):
24 self.response = None
25 self.corr_id = str(uuid.uuid4()) #惟一标识符
26 self.channel.basic_publish(exchange='',
27 routing_key='rpc_queue3', #声明一个Q
28 properties=pika.BasicProperties(
29 reply_to=self.callback_queue,
30 correlation_id=self.corr_id,
31 ),
32 body=str(n))
33
34 print("start waiting for cmd result ")
35 count = 0
36 while self.response is None: #若是命令没返回结果
37 print("loop ",count)
38 count +=1
39 self.connection.process_data_events() #以不阻塞的形式去检测有没有新事件
40 #若是没事件，那就什么也不作， 若是有事件，就触发on_response事件
41 return self.response
42
43 ssh_rpc = SSHRpcClient()
44
45 print(" [x] sending cmd")
46 response = ssh_rpc.call("ipconfig")
47
48 print(" [.] Got result ")
49 print(response.decode("gbk"))
rabbit_rpc_recv.py（消费端是接收端）
1 import pika
2 import time
3 import subprocess
4
5 credentials = pika.PlainCredentials('nulige', '123456')
6 connection = pika.BlockingConnection(pika.ConnectionParameters(
7 host='192.168.1.118', credentials=credentials))
8
9 channel = connection.channel()
10 channel.queue_declare(queue='rpc_queue3')
11
12 def SSHRPCServer(cmd):
13
14 print("recv cmd:",cmd)
15 cmd_obj = subprocess.Popen(cmd.decode(),shell=True,stdout=subprocess.PIPE,stderr=subprocess.PIPE)
16
17 result = cmd_obj.stdout.read() or cmd_obj.stderr.read()
18 return result
19
20 def on_request(ch, method, props, body):
21
22 print(" [.] fib(%s)" % body)
23 response = SSHRPCServer(body)
24
25 ch.basic_publish(exchange='',
26 routing_key=props.reply_to,
27 properties=pika.BasicProperties(correlation_id= \
28 props.correlation_id),
29 body=response)
30
31 channel.basic_consume(on_request, queue='rpc_queue3')
32 print(" [x] Awaiting RPC requests")
33 channel.start_consuming()
相关文章
- 1. 缓存、队列（Memcached、redis、RabbitMQ）
- 2. 缓存，队列（Redis，RabbitMQ）
- 3. 缓存、队列（Memcached，Redis，rabbitMQ）
- 4. redis缓存,rabbitMQ队列
- 5. WEEK11：redis缓存、rabbitMQ队列
- 6. 缓存, 队列(Redis,RabbitMQ)
- 7. Redis实现缓存队列
- 8. day10-python-协程\异步IO\RabbitMQ队列\redis缓存
- 9. Day10 - Python协程、异步IO、redis缓存、rabbitMQ队列
- 10. Python 10--协程、异步IO、redis缓存、rabbitMQ队列
- 更多相关文章...
- • C# 队列（Queue） - C#教程
- • Hibernate的二级缓存 - Hibernate教程
- • 三篇文章了解 TiDB 技术内幕——说存储
- • 算法总结-滑动窗口
相关标签/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
欢迎关注本站公众号,获取更多信息
相关文章
- 1. 缓存、队列（Memcached、redis、RabbitMQ）
- 2. 缓存，队列（Redis，RabbitMQ）
- 3. 缓存、队列（Memcached，Redis，rabbitMQ）
- 4. redis缓存,rabbitMQ队列
- 5. WEEK11：redis缓存、rabbitMQ队列
- 6. 缓存, 队列(Redis,RabbitMQ)
- 7. Redis实现缓存队列
- 8. day10-python-协程\异步IO\RabbitMQ队列\redis缓存
- 9. Day10 - Python协程、异步IO、redis缓存、rabbitMQ队列
- 10. Python 10--协程、异步IO、redis缓存、rabbitMQ队列