安装Rabbitmq，通过Rabbitmq实现RPC全面了解

　　Rabbitmq一：消息队列介绍1。介绍消息队列就是基础数据结构中的先进先出的一种数据机构。想一下，生活中买东西需要排队，先排队的人先买消费，就是典型的先进先出。
　　扩展redis：可以作为简单的消息队列celery：本事就是基于消息队列进行的封装。2。MQ解决了什么问题MQ是一直存在，不过随着微服务架构的流行，成了解决微服务和微服务之间通信的常用工具。扩展1。两个服务之间调用的方式：1。restful七层协议oss（http协议）2。rpctcpsocket层（远程过程调用）2。不管是使用restful还是rpc，都涉及到使用同步还是异步：1。异步：client使用rpc和server交互，client使用异步，不管有没有执行成功，就不停的异步的提交数据，数据在server消息队列排着队，等待着消费。1。应用的解耦1。以电商应用为例，应用中有订单系统，库存系统，物流系统，支付系统。用户创建订单后，如果耦合调用库存系统，物流系统，支付系统，任何一个子系统出现了故障，都会造成下单操作异常。2。当转变成基于队列的方式后，系统间调用的问题会减少很多，比如物流系统因为发生故障，需要几分钟来修复。在这几分钟的时间里，物流系统要处理的内存被缓存在消息队列中，用户的下单操作可以正常完成。当物流系统恢复后，继续处理订单信息即可，订单用户感受不到物流系统的故障。提升系统的可用性。
　　2。流量削峰1。举个列子，如果订单系统最多能处理一万次订单，这个处理能力应付正常时段的下单时绰绰有余，正常时段我们下单一秒后就能返回结果。但是在高峰期，如果有两万次下单操作系统是处理不了的，只能限制订单超过一万后不允许用户下单。2。使用消息队列做缓存，我们可以取消这个限制，把一秒内下的订单分散成一段时间来处理，这时有些用户可能在下单十几秒后才能收到下单成功的操作，但是比不能下单的体验要好。结：1。通常下比如有两万订单，这时我们server肯定消费不过来，我们将两万丢到消息队列中，进行消费即可。就叫流量消峰如：双十一，消息队列多消费3。消息分发（发布订阅：观察者模式）多个服务对数据感兴趣，只需要监听同一类消息即可处理。
　　列如A产生数据，B对数据感兴趣。如果没有消息队列A每次处理完需要调用一下B服务。过了一段时间C对数据也感兴趣，A就需要改代码，调用B服务，调用C服务。只要有服务需要，A服务都要改动代码。很不方便。xxxxxxxxxx有了消息队列后，A只管发送一次消息，B对消息感兴趣，只需要监听消息。C感兴趣，C也去监听消息。A服务作为基础服务完全不需要有改动。4。异步消息（celery就是对消息队列的封装）xxxxxxxxxx有些服务间调用是异步的：1。例如A调用B，B需要花费很长时间执行，但是A需要知道B什么时候可以执行完，以前一般有两种方式，A过了一段时间去调用B的查询api是否完成。2。或者A提供一个callbackapi，B执行完之后调用api通知A服务。这两种方式都不是很优雅。python
　　1。使用消息总线，可以很方便解决这个问题，A调用B服务后，只需要监听B处理完成的消息，当B处理完成后，会发送一条消息给MQ，MQ会将此消息转发给A服务。2。这样A服务既不用循环调用B的查询api，也不用提供callbackapi。同样B服务也不用做这些操作。A服务还能及时的得到异步处理成功的消息。3。常见消息队列及比较xxxxxxxxxxRabbitMQ：支持持久化，断电后，重启，数据是不会丢的。1。吞吐量小：几百万都是没问题的，消息确认：我告诉你，我消费完了，你在删2。应用场景：订单，对消息可靠性有要求，就用它Kafka：吞吐量高，注重高吞吐量，不注重消息的可靠性1。你拿走就没了，消费过程崩了，就没了。2。应用场景，数据量特别大。结论：1。Kafka在于分布式架构，RabbitMQ基于AMQP协议来实现，RocketMQ思路来源于Kafka，改成了主从结构，在事务性可靠性方面做了优化。2。广泛来说，电商，金融等对事物性要求很高的，可以考虑RabbitMQ，对性能要求或吞吐量高的可考虑Kafka。python二：Rabbitmq安装安装两种官网：https：www。rabbitmq。comgetstarted。htmldockerhub下载指定的rabbitmq：management的RabbitMQ
　　1。服务端原生安装1原生安装安装扩展epel源yumyinstallerlangyumyinstallrabbitmqserver查询是否安装rpmqarabbitmqserversystemctlstartrabbitmqserver以上也有web管理页面，只不过需要配置文件配置。第一种方式客户端连接服务端，可以不用配置用户和密码，只需要ip连接。第二种方式则需要配置用户名和密码。2。服务端docker拉取2docker拉取dockerpullrabbitmq：management（自动开启了web管理界面）dockerrundinamerabbitmqeRABBITMQDEFAULTUSERadmineRABBITMQDEFAULTPASSadminp15672：15672p5672：5672rabbitmq：management直接run如果没有rabbitmq就会自动拉dockerrundiname：指定rabbitmqe：环境变量eRABBITMQDEFAULTUSERadmin：用户名eRABBITMQDEFAULTPASSadmin：密码p15672：15672：rabbitmqweb管理界面端口p5672：5672：rabbitmq默认的监听端口
　　dockerps
　　http：47。101。159。222：15672
　　3。Rabbitmq可视化界面创建用户（设置用户和密码）
　　4。命令创建Rabbitmq用户（设置用户和密码）4创建用户rabbitmqctladduserlqz1235分配权限设置用户为admin角色rabbitmqctlsetusertagslqzadministrator设置权限rabbitmqctlsetpermissionsplqz。。。rabbitmqctlsetpermissionsp用户名。。。三：客户端安装pip3installpika四：基本使用（生产者消费者模型）对于RabbitMQ来说，生产和消费不再针对内存里的一个Queue对象，而是某台服务器上的RabbitMQServer实现的消息队列。生产者importpika创建连接对象connectionpika。BlockingConnection（pika。ConnectionParameters（host47。101。159。222：22））host指定rabbitmq服务器ip地址指定用户名和密码，如果rabbitmq没有设置用户名和密码，可以不指定credentialspika。PlainCredentials（admin，admin）connectionpika。BlockingConnection（pika。ConnectionParameters（47。101。159。222，credentialscredentials））host指定rabbitmq服务器ip地址，credentials指定用户名和密码创建channel对象，用于发送消息，接收消息，声明队列channelconnection。channel（）connection。channel（）：在连接上创建一个频道，这个频道就是我们要通过它来发送，接收消息的对象，类似于TCP中的socket对象，我们通过它来收发消息，声明队列声明队列，如果队列不存在，则创建队列，如果队列存在，则不创建channel。queuedeclare（queuedatalog）生产者向队列中放一条消息channel。basicpublish（exchange，交换机，如果不指定，则使用默认的交换机，默认的交换机routingkeydatalog，队列名称bodyzllnb！发送的消息）print（SentHelloWorld！）关闭连接connection。close（）消费者importpika，sys，osdefmain（）：创建连接对象connectionpika。BlockingConnection（pika。ConnectionParameters（host47。101。159。222：22））host指定rabbitmq服务器ip地址指定用户名和密码，如果rabbitmq没有设置用户名和密码，可以不指定credentialspika。PlainCredentials（admin，admin）connectionpika。BlockingConnection（pika。ConnectionParameters（47。101。159。222，credentialscredentials））host指定rabbitmq服务器ip地址，credentials指定用户名和密码创建channel对象，用于发送消息，接收消息，声明队列channelconnection。channel（）connection。channel（）：在连接上创建一个频道，这个频道就是我们要通过它来发送，接收消息的对象，类似于TCP中的socket对象，我们通过它来收发消息，声明队列消费者也声明了队列，因为如果是消费者先启动，那么队列就不存在了，消费者就无法消费消息了，所以消费者也要声明队列channel。queuedeclare（queuedatalog）defcallback（ch，method，properties，body）：print（〔x〕Receivedrbody）消费者从队列queue指定的消费队列hello中取消息，拿到数据了之前将hello队列的数据丢到callback里面，如果队列中没有消息，则会一直等待，直到有消息为止autoackTrue：自动确认消息，如果不设置为True，那么消息会一直处于未确认状态，即使消费者已经消费了消息，消息也不会从队列中删除，这样就会造成消息的重复消费，所以一般都会设置为Trueautoacktrue：队列接收到既直接确认，就会删除队列中的消息，不会管后面数据会不会消费完。channel。basicconsume（queuedatalog，onmessagecallbackcallback，autoackTrue）默认为false，不自动确认消息，需要手动确认print（〔〕Waitingformessages。ToexitpressCTRLC）channel。startconsuming（）开始消费消息，如果队列中没有消息，那么就会一直等待，直到有消息为止，如果队列中有消息，那么就会消费消息ifnamemain：main（）
　　五：消息确认机制（消息安全之ack）autoack：自动确认消息（队列接收到就会确认消费，会丢失数据的可能性）默认为falseautoacktrue：队列接收到既直接确认，就会删除队列中的消息，不会管后面数据会不会消费完。autoackfalse：设置为false的情况，那么消息会一直处于未确认状态，即使消费者已经消费了消息，消息也不会从队列中删除，这样就会造成消息的重复消费ch。basicack：消费完后，自动确认消费（可靠性，保证数据都完整的消费）：常用推荐ch。basicack（deliverytagmethod。deliverytag）：真正的将消息消费完了后，再发确认，就会删除掉队列中的消息。生产者importpika创建连接对象connectionpika。BlockingConnection（pika。ConnectionParameters（host47。101。159。222：22））host指定rabbitmq服务器ip地址指定用户名和密码，如果rabbitmq没有设置用户名和密码，可以不指定credentialspika。PlainCredentials（admin，admin）connectionpika。BlockingConnection（pika。ConnectionParameters（47。101。159。222，credentialscredentials））host指定rabbitmq服务器ip地址，credentials指定用户名和密码创建channel对象，用于发送消息，接收消息，声明队列channelconnection。channel（）connection。channel（）：在连接上创建一个频道，这个频道就是我们要通过它来发送，接收消息的对象，类似于TCP中的socket对象，我们通过它来收发消息，声明队列声明队列，如果队列不存在，则创建队列，如果队列存在，则不创建channel。queuedeclare（queuedatalog）生产者向队列中放一条消息channel。basicpublish（exchange，交换机，如果不指定，则使用默认的交换机，默认的交换机routingkeydatalog，队列名称bodyzllnb！发送的消息）print（SentHelloWorld！）关闭连接connection。close（）消费者importpika，sys，osdefmain（）：创建连接对象connectionpika。BlockingConnection（pika。ConnectionParameters（host47。101。159。222：22））host指定rabbitmq服务器ip地址指定用户名和密码，如果rabbitmq没有设置用户名和密码，可以不指定credentialspika。PlainCredentials（admin，admin）connectionpika。BlockingConnection（pika。ConnectionParameters（47。101。159。222，credentialscredentials））host指定rabbitmq服务器ip地址，credentials指定用户名和密码创建channel对象，用于发送消息，接收消息，声明队列channelconnection。channel（）connection。channel（）：在连接上创建一个频道，这个频道就是我们要通过它来发送，接收消息的对象，类似于TCP中的socket对象，我们通过它来收发消息，声明队列消费者也声明了队列，因为如果是消费者先启动，那么队列就不存在了，消费者就无法消费消息了，所以消费者也要声明队列channel。queuedeclare（queuedatalog）defcallback（ch，method，properties，body）：print（〔x〕Receivedrbody）真正的将消息消费完了，再发确认ch。basicack（deliverytagmethod。deliverytag）消费者从队列queue指定的消费队列hello中取消息，拿到数据了之前将hello队列的数据丢到callback里面，如果队列中没有消息，则会一直等待，直到有消息为止autoackTrue：自动确认消息，如果不设置为True，那么消息会一直处于未确认状态，即使消费者已经消费了消息，消息也不会从队列中删除，这样就会造成消息的重复消费，所以一般都会设置为Trueautoacktrue：队列接收到既直接确认，就会删除队列中的消息，不会管后面数据会不会消费完。channel。basicconsume（queuedatalog，onmessagecallbackcallback，autoackFalse）默认为false，不自动确认消息，需要手动确认print（〔〕Waitingformessages。ToexitpressCTRLC）channel。startconsuming（）开始消费消息，如果队列中没有消息，那么就会一直等待，直到有消息为止，如果队列中有消息，那么就会消费消息ifnamemain：main（）六：持久化（消息安全之durable持久化）1。什么是rabbitmq持久化？数据支持持久化，运行过程中，rabbitmq宕机了，在重新启动起来，如果队列消费消息没被消费，那么就还是会存在。2。配置队列持久化在创建队列的时候增加durableTrue设置队列持久化，如果rabbitmq服务重启，队列不会丢失channel。queuedeclare（queuedatalog，durableTrue）3。配置消息持久化在发布消息的时候增加properties设置消息持久化，如果rabbitmq服务停止，重启后，消息还在，1：非持久化，2：持久化，默认为1propertiespika。BasicProperties（deliverymode2，）注意：1。没加持久化配置之前的队列不会支持持久化，需要加持久化配置之后重新创建。生产者importpika创建连接对象connectionpika。BlockingConnection（pika。ConnectionParameters（host47。101。159。222：22））host指定rabbitmq服务器ip地址指定用户名和密码，如果rabbitmq没有设置用户名和密码，可以不指定credentialspika。PlainCredentials（admin，admin）connectionpika。BlockingConnection（pika。ConnectionParameters（47。101。159。222，credentialscredentials））host指定rabbitmq服务器ip地址，credentials指定用户名和密码创建channel对象，用于发送消息，接收消息，声明队列channelconnection。channel（）connection。channel（）：在连接上创建一个频道，这个频道就是我们要通过它来发送，接收消息的对象，类似于TCP中的socket对象，我们通过它来收发消息，声明队列声明队列，如果队列不存在，则创建队列，如果队列存在，则不创建channel。queuedeclare（queuedatalog，durableTrue）durableTrue：队列持久化，如果rabbitmq服务停止，重启后，队列还在生产者向队列中放一条消息channel。basicpublish（exchange，交换机，如果不指定，则使用默认的交换机，默认的交换机routingkeydatalog，队列名称bodyzllnb！，发送的消息propertiespika。BasicProperties（deliverymode2，消息持久化，如果rabbitmq服务停止，重启后，消息还在，1：非持久化，2：持久化））print（SentHelloWorld！）关闭连接connection。close（）消费者importpika，sys，osdefmain（）：创建连接对象connectionpika。BlockingConnection（pika。ConnectionParameters（host47。101。159。222：22））host指定rabbitmq服务器ip地址指定用户名和密码，如果rabbitmq没有设置用户名和密码，可以不指定credentialspika。PlainCredentials（admin，admin）connectionpika。BlockingConnection（pika。ConnectionParameters（47。101。159。222，credentialscredentials））host指定rabbitmq服务器ip地址，credentials指定用户名和密码创建channel对象，用于发送消息，接收消息，声明队列channelconnection。channel（）connection。channel（）：在连接上创建一个频道，这个频道就是我们要通过它来发送，接收消息的对象，类似于TCP中的socket对象，我们通过它来收发消息，声明队列消费者也声明了队列，因为如果是消费者先启动，那么队列就不存在了，消费者就无法消费消息了，所以消费者也要声明队列channel。queuedeclare（queuedatalog，durableTrue）durableTrue：队列持久化，如果rabbitmq服务重启，队列不会丢失defcallback（ch，method，properties，body）：print（〔x〕Receivedrbody）真正的将消息消费完了，再发确认ch。basicack（deliverytagmethod。deliverytag）消费者从队列queue指定的消费队列hello中取消息，拿到数据了之前将hello队列的数据丢到callback里面，如果队列中没有消息，则会一直等待，直到有消息为止autoackTrue：自动确认消息，如果不设置为True，那么消息会一直处于未确认状态，即使消费者已经消费了消息，消息也不会从队列中删除，这样就会造成消息的重复消费，所以一般都会设置为Trueautoacktrue：队列接收到既直接确认，就会删除队列中的消息，不会管后面数据会不会消费完。channel。basicconsume（queuedatalog，onmessagecallbackcallback，autoackFalse）默认为false，不自动确认消息，需要手动确认print（〔〕Waitingformessages。ToexitpressCTRLC）channel。startconsuming（）开始消费消息，如果队列中没有消息，那么就会一直等待，直到有消息为止，如果队列中有消息，那么就会消费消息ifnamemain：main（）
　　七：闲置消费1。什么是闲置消费？1。正常情况如果有多个消费者，是按照顺序第一个消息给第一个消费者，第二个消息给第二个消费者，以此类推，只能按照顺序。2。但是可能第一个消费的消费者处理消息很耗时，一直没结束，此时就可以让第二个消费者优先获取闲置的消息，次方法就称之为闲置消费。2。配置闲置消费消费者配置，每次只接收一条消息，处理完了再接收下一条，这样可以保证消息的顺序性，不会出现消息乱序的情况channel。basicqos（prefetchcount1）1代表每次只接收一条消息，接收完了再接收下一条缺点：1。但是会降低效率，因为每次只处理一条消息，如果消息处理很快，那么效率就会降低生产者importpika创建连接对象connectionpika。BlockingConnection（pika。ConnectionParameters（host47。101。159。222：22））host指定rabbitmq服务器ip地址指定用户名和密码，如果rabbitmq没有设置用户名和密码，可以不指定credentialspika。PlainCredentials（admin，admin）connectionpika。BlockingConnection（pika。ConnectionParameters（47。101。159。222，credentialscredentials））host指定rabbitmq服务器ip地址，credentials指定用户名和密码创建channel对象，用于发送消息，接收消息，声明队列channelconnection。channel（）connection。channel（）：在连接上创建一个频道，这个频道就是我们要通过它来发送，接收消息的对象，类似于TCP中的socket对象，我们通过它来收发消息，声明队列声明队列，如果队列不存在，则创建队列，如果队列存在，则不创建channel。queuedeclare（queuedatalog，durableTrue）durableTrue：队列持久化，如果rabbitmq服务停止，重启后，队列还在生产者向队列中放一条消息channel。basicpublish（exchange，交换机，如果不指定，则使用默认的交换机，默认的交换机routingkeydatalog，队列名称bodyzllnb！，发送的消息propertiespika。BasicProperties（deliverymode2，消息持久化，如果rabbitmq服务停止，重启后，消息还在，1：非持久化，2：持久化））print（SentHelloWorld！）关闭连接connection。close（）消费者1importtimeimportpika，sys，osdefmain（）：创建连接对象connectionpika。BlockingConnection（pika。ConnectionParameters（host47。101。159。222：22））host指定rabbitmq服务器ip地址指定用户名和密码，如果rabbitmq没有设置用户名和密码，可以不指定credentialspika。PlainCredentials（admin，admin）connectionpika。BlockingConnection（pika。ConnectionParameters（47。101。159。222，credentialscredentials））host指定rabbitmq服务器ip地址，credentials指定用户名和密码创建channel对象，用于发送消息，接收消息，声明队列channelconnection。channel（）connection。channel（）：在连接上创建一个频道，这个频道就是我们要通过它来发送，接收消息的对象，类似于TCP中的socket对象，我们通过它来收发消息，声明队列消费者也声明了队列，因为如果是消费者先启动，那么队列就不存在了，消费者就无法消费消息了，所以消费者也要声明队列channel。queuedeclare（queuedatalog，durableTrue）durableTrue：队列持久化，如果rabbitmq服务重启，队列不会丢失defcallback（ch，method，properties，body）：print（〔x〕Receivedrbody）time。sleep（50）模拟处理任务，耗时50秒真正的将消息消费完了，再发确认ch。basicack（deliverytagmethod。deliverytag）闲置消费channel。basicqos（prefetchcount1）每次只接收一条消息，处理完了再接收下一条，这样可以保证消息的顺序性，不会出现消息乱序的情况，但是会降低效率，因为每次只处理一条消息，如果消息处理很快，那么效率就会降低消费者从队列queue指定的消费队列hello中取消息，拿到数据了之前将hello队列的数据丢到callback里面，如果队列中没有消息，则会一直等待，直到有消息为止autoackTrue：自动确认消息，如果不设置为True，那么消息会一直处于未确认状态，即使消费者已经消费了消息，消息也不会从队列中删除，这样就会造成消息的重复消费，所以一般都会设置为Trueautoacktrue：队列接收到既直接确认，就会删除队列中的消息，不会管后面数据会不会消费完。channel。basicconsume（queuedatalog，onmessagecallbackcallback，autoackFalse）默认为false，不自动确认消息，需要手动确认print（〔〕Waitingformessages。ToexitpressCTRLC）channel。startconsuming（）开始消费消息，如果队列中没有消息，那么就会一直等待，直到有消息为止，如果队列中有消息，那么就会消费消息ifnamemain：main（）消费者2importpika，sys，osdefmain（）：创建连接对象connectionpika。BlockingConnection（pika。ConnectionParameters（host47。101。159。222：22））host指定rabbitmq服务器ip地址指定用户名和密码，如果rabbitmq没有设置用户名和密码，可以不指定credentialspika。PlainCredentials（admin，admin）connectionpika。BlockingConnection（pika。ConnectionParameters（47。101。159。222，credentialscredentials））host指定rabbitmq服务器ip地址，credentials指定用户名和密码创建channel对象，用于发送消息，接收消息，声明队列channelconnection。channel（）connection。channel（）：在连接上创建一个频道，这个频道就是我们要通过它来发送，接收消息的对象，类似于TCP中的socket对象，我们通过它来收发消息，声明队列消费者也声明了队列，因为如果是消费者先启动，那么队列就不存在了，消费者就无法消费消息了，所以消费者也要声明队列channel。queuedeclare（queuedatalog，durableTrue）durableTrue：队列持久化，如果rabbitmq服务重启，队列不会丢失defcallback（ch，method，properties，body）：print（〔x〕Receivedrbody）真正的将消息消费完了，再发确认ch。basicack（deliverytagmethod。deliverytag）闲置消费channel。basicqos（prefetchcount1）每次只接收一条消息，处理完了再接收下一条，这样可以保证消息的顺序性，不会出现消息乱序的情况，但是会降低效率，因为每次只处理一条消息，如果消息处理很快，那么效率就会降低消费者从队列queue指定的消费队列hello中取消息，拿到数据了之前将hello队列的数据丢到callback里面，如果队列中没有消息，则会一直等待，直到有消息为止autoackTrue：自动确认消息，如果不设置为True，那么消息会一直处于未确认状态，即使消费者已经消费了消息，消息也不会从队列中删除，这样就会造成消息的重复消费，所以一般都会设置为Trueautoacktrue：队列接收到既直接确认，就会删除队列中的消息，不会管后面数据会不会消费完。channel。basicconsume（queuedatalog，onmessagecallbackcallback，autoackFalse）默认为false，不自动确认消息，需要手动确认print（〔〕Waitingformessages。ToexitpressCTRLC）channel。startconsuming（）开始消费消息，如果队列中没有消息，那么就会一直等待，直到有消息为止，如果队列中有消息，那么就会消费消息ifnamemain：main（）
　　八：发布订阅（fanout）发布订阅：可以有多个订阅者来订阅发布者的消息fanout：不需要routingkey，只需要将消息发送到交换机中，交换机会将消息发送到所有绑定到它的队列中实现发布订阅逻辑1。发布者P将消息发送到X交换机上面，2。C1，C2，多个订阅者随机创建出多个队列，将订阅者队列绑定给X交换机，3。X交换机通过队列将数据发送给所有绑定X交换机的订阅者。
　　发布订阅生产者importpika指定用户名和密码，如果rabbitmq没有设置用户名和密码，可以不指定credentialspika。PlainCredentials（admin，admin）host指定rabbitmq服务器ip地址，credentials指定用户名和密码connectionpika。BlockingConnection（pika。ConnectionParameters（47。101。159。222，credentialscredentials））在连接上创建一个频道，这个频道就是我们要通过它来发送，接收消息的对象，类似于TCP中的socket对象，我们通过它来收发消息，声明队列channelconnection。channel（）不需要声明队列，因为生产者不需要将消息放到队列中，只需要将消息发送到交换机中即可channel。exchangedeclare（exchangelogs，exchangetypefanout）声明交换机，交换机类型为fanoutexchangetype的三种类型：1、direct：根据routingkey将消息放到对应的队列中2、topic：根据routingkey和bindingkey将消息放到对应的队列中3、fanout：不需要routingkey，只需要将消息发送到交换机中，交换机会将消息发送到所有绑定到它的队列中messageinfo：HelloWorld！发送消息到交换机中channel。basicpublish（exchangelogs，routingkey，bodymessage）发送消息到交换机connection。close（）订阅者消费者importpika指定用户名和密码，如果rabbitmq没有设置用户名和密码，可以不指定credentialspika。PlainCredentials（admin，admin）host指定rabbitmq服务器ip地址，credentials指定用户名和密码connectionpika。BlockingConnection（pika。ConnectionParameters（47。101。159。222，credentialscredentials））在连接上创建一个频道，这个频道就是我们要通过它来发送，接收消息的对象，类似于TCP中的socket对象，我们通过它来收发消息，声明队列channelconnection。channel（）不需要声明队列，因为生产者不需要将消息放到队列中，只需要将消息发送到交换机中即可channel。exchangedeclare（exchangelogs，exchangetypefanout）声明交换机，交换机类型为fanout声明一个随机队列，exclusiveTrue表示这个队列只能被当前连接使用，当连接关闭时，队列会被删除，exclusiveTrue是为了防止多个消费者同时消费一个队列，导致消息重复消费resultchannel。queuedeclare（queue，exclusiveTrue）获取随机队列的名称，随机的意义是什么：每次运行程序都会创建一个新的队列，这样就不会有多个消费者同时消费同一个队列中的消息，这样就可以实现消息的负载均衡，每个消费者都会平均的消费队列中的消息queuenameresult。method。queue默认会创建一个随机队列，队列名称是随机的。这个队列只能被当前连接使用，当连接关闭时，队列会被删除。也可以指定队列名称，但是要确保队列名称是唯一的，不然会报错print（queuename）channel。queuebind（exchangelogs，queuequeuename）将队列绑定到交换机上，交换机类型为fanout，所以不需要指定routingkey，交换机会将消息发送到所有绑定到它上面的队列print（〔〕Waitingforlogs。ToexitpressCTRLC）defcallback（ch，method，properties，body）：print（〔x〕rbody）channel。basicconsume（queuequeuename，onmessagecallbackcallback，autoackTrue）channel。startconsuming（）
　　九：关键字（direct）direct：根据routingkey将消息放到对应的队列中1。关键字1。将随机队列绑定到交换机上，routingkey指定路由键，这里指定的是key，2。表示只有路由键为info的消息才会被发送到该随机队列中，也就是说只有生产者发送的消息的路由键为key的消息才会被消费。总结：将随机队列绑定到交换机上，routingkey为指定消费交换机的队列名称，从而实现指定消费，然后将消息从绑定的交换机的队列获取，消费。routingkey监听的队列名称发布订阅生产者importpika指定用户名和密码，如果rabbitmq没有设置用户名和密码，可以不指定credentialspika。PlainCredentials（admin，admin）host指定rabbitmq服务器ip地址，credentials指定用户名和密码connectionpika。BlockingConnection（pika。ConnectionParameters（47。101。159。222，credentialscredentials））在连接上创建一个频道，这个频道就是我们要通过它来发送，接收消息的对象，类似于TCP中的socket对象，我们通过它来收发消息，声明队列channelconnection。channel（）不需要声明队列，因为生产者不需要将消息放到队列中，只需要将消息发送到交换机中即可channel。exchangedeclare（exchangezll，exchangetypedirect）声明交换机，交换机类型为directexchangetype的三种类型：1、direct：根据routingkey将消息放到对应的队列中2、topic：根据routingkey和bindingkey将消息放到对应的队列中3、fanout：不需要routingkey，只需要将消息发送到交换机中，交换机会将消息发送到所有绑定到它的队列中messageinfo：HelloWorld！发送消息到交换机中channel。basicpublish（exchangezll，routingkeybnb，bodymessage）routingkey为bnb，消息会被发送到bnb队列中，如果没有bnb队列，消息会被丢弃，因为没有队列可以接收消息connection。close（）订阅者消费者importpika指定用户名和密码，如果rabbitmq没有设置用户名和密码，可以不指定credentialspika。PlainCredentials（admin，admin）host指定rabbitmq服务器ip地址，credentials指定用户名和密码connectionpika。BlockingConnection（pika。ConnectionParameters（47。101。159。222，credentialscredentials））在连接上创建一个频道，这个频道就是我们要通过它来发送，接收消息的对象，类似于TCP中的socket对象，我们通过它来收发消息，声明队列channelconnection。channel（）不需要声明队列，因为生产者不需要将消息放到队列中，只需要将消息发送到交换机中即可channel。exchangedeclare（exchangezll，exchangetypedirect）声明交换机，交换机类型为fanout声明一个随机队列，exclusiveTrue表示这个队列只能被当前连接使用，当连接关闭时，队列会被删除，exclusiveTrue是为了防止多个消费者同时消费一个队列，导致消息重复消费resultchannel。queuedeclare（queue，exclusiveTrue）获取随机队列的名称，随机的意义是什么：每次运行程序都会创建一个新的队列，这样就不会有多个消费者同时消费同一个队列中的消息，这样就可以实现消息的负载均衡，每个消费者都会平均的消费队列中的消息queuenameresult。method。queue默认会创建一个随机队列，队列名称是随机的。这个队列只能被当前连接使用，当连接关闭时，队列会被删除。也可以指定队列名称，但是要确保队列名称是唯一的，不然会报错print（queuename）将队列绑定到交换机上，routingkey指定路由键，这里指定的是info，表示只有路由键为info的消息才会被发送到该随机队列中，也就是说只有生产者发送的消息的路由键为info的消息才会被消费。channel。queuebind（exchangezll，queuequeuename，routingkeynb）将队列绑定到交换机上，routingkey为指定消费交换机的队列名称，从而实现指定消费，然后将消息从绑定到交换机的队列获取channel。queuebind（exchangezll，queuequeuename，routingkeybnb）print（〔〕Waitingforlogs。ToexitpressCTRLC）defcallback（ch，method，properties，body）：print（〔x〕rbody）channel。basicconsume（queuequeuename，onmessagecallbackcallback，autoackTrue）channel。startconsuming（）十：模糊匹配（topic）topic：根据routingkey和bindingkey将消息放到对应的队列中模糊匹配的关键：表示后面可以跟任意字符：表示后面只能跟一个单词发布订阅生产者importpika指定用户名和密码，如果rabbitmq没有设置用户名和密码，可以不指定credentialspika。PlainCredentials（admin，admin）host指定rabbitmq服务器ip地址，credentials指定用户名和密码connectionpika。BlockingConnection（pika。ConnectionParameters（47。101。159。222，credentialscredentials））在连接上创建一个频道，这个频道就是我们要通过它来发送，接收消息的对象，类似于TCP中的socket对象，我们通过它来收发消息，声明队列channelconnection。channel（）不需要声明队列，因为生产者不需要将消息放到队列中，只需要将消息发送到交换机中即可channel。exchangedeclare（exchangeaaa，exchangetypetopic）声明交换机，交换机类型为directexchangetype的三种类型：1、direct：根据routingkey将消息放到对应的队列中2、topic：根据routingkey和bindingkey将消息放到对应的队列中3、fanout：不需要routingkey，只需要将消息发送到交换机中，交换机会将消息发送到所有绑定到它的队列中messageinfo：HelloWorld！发送消息到交换机中channel。basicpublish（exchangeaaa，routingkeybnb。xxxx，bodymessage）routingkey为bnb，消息会被发送到bnb队列中，如果没有bnb队列，消息会被丢弃，因为没有队列可以接收消息connection。close（）消费者importpika指定用户名和密码，如果rabbitmq没有设置用户名和密码，可以不指定credentialspika。PlainCredentials（admin，admin）host指定rabbitmq服务器ip地址，credentials指定用户名和密码connectionpika。BlockingConnection（pika。ConnectionParameters（47。101。159。222，credentialscredentials））在连接上创建一个频道，这个频道就是我们要通过它来发送，接收消息的对象，类似于TCP中的socket对象，我们通过它来收发消息，声明队列channelconnection。channel（）不需要声明队列，因为生产者不需要将消息放到队列中，只需要将消息发送到交换机中即可channel。exchangedeclare（exchangeaaa，exchangetypetopic）声明交换机，交换机类型为fanout声明一个随机队列，exclusiveTrue表示这个队列只能被当前连接使用，当连接关闭时，队列会被删除，exclusiveTrue是为了防止多个消费者同时消费一个队列，导致消息重复消费resultchannel。queuedeclare（queue，exclusiveTrue）获取随机队列的名称，随机的意义是什么：每次运行程序都会创建一个新的队列，这样就不会有多个消费者同时消费同一个队列中的消息，这样就可以实现消息的负载均衡，每个消费者都会平均的消费队列中的消息queuenameresult。method。queue默认会创建一个随机队列，队列名称是随机的。这个队列只能被当前连接使用，当连接关闭时，队列会被删除。也可以指定队列名称，但是要确保队列名称是唯一的，不然会报错print（queuename）将随机队列绑定到交换机上，routingkey为指定消费交换机的队列名称，从而实现指定消费，然后将消息从绑定的交换机的队列获取，routingkey监听的队列名称channel。queuebind（exchangeaaa，queuequeuename，routingkeynb）将队列绑定到交换机上，routingkey为指定消费交换机的队列名称，从而实现指定消费，然后将消息从绑定到交换机的队列获取channel。queuebind（exchangeaaa，queuequeuename，routingkeybnb。）print（〔〕Waitingforlogs。ToexitpressCTRLC）defcallback（ch，method，properties，body）：print（〔x〕rbody）channel。basicconsume（queuequeuename，onmessagecallbackcallback，autoackTrue）channel。startconsuming（）
　　十一：通过rabbitmq实现rpc（基于RabbitMQ封装RPC）通过RabbitMQ实现rpc实现逻辑1。服务端启动接收消息，监听queue队列。2。实列化客户端，调用call方法，将消息属性内包含：1。回调函数随机队列，接收服务端返回结果，服务端会将结果发送到这个队列。2。客户但的随机uuid，标识唯一消息。然后将body消息发送给服务端。3。客户端，发布完消息后，进入非阻塞状态，如果没有接收到服务端返回的结果，会一直等待，直到收到结果，然后返回结果。4。服务端接收queue队列消息，调用函数将消息进行处理，获取裴波那契数列。5。然后服务端进行发布，将消息发送到客户端的回调函数队列，客户端的uuid。6。客户端监听接收队列消息，调用函数处理，判断唯一uuid，确认body，然后成功收到消息并返回。服务端importpika指定用户名和密码，如果rabbitmq没有设置用户名和密码，可以不指定credentialspika。PlainCredentials（admin，admin）host指定rabbitmq服务器ip地址，credentials指定用户名和密码connectionpika。BlockingConnection（pika。ConnectionParameters（47。101。159。222，credentialscredentials））在连接上创建一个频道，这个频道就是我们要通过它来发送，接收消息的对象，类似于TCP中的socket对象，我们通过它来收发消息，声明队列channelconnection。channel（）channel。queuedeclare（queuerpcqueue）声明队列，如果队列不存在，会自动创建deffib（n）：ifn0：return0elifn1：return1else：递归调用，计算斐波那契数列returnfib（n1）fib（n2）defonrequest（ch，method，props，body）：ch为频道，method为方法，props为属性，body为消息体nint（body）print（〔。〕fib（s）n）responsefib（n）ch。basicpublish（exchange，routingkeyprops。replyto，将消息发送到客户端的回调函数propertiespika。BasicProperties（correlationidprops。correlationid），将客户端的correlationid传递给客户端bodystr（response））发送ack消息，告诉rabbitmq，消息已经被处理ch。basicack（deliverytagmethod。deliverytag）每次只接收一个消息channel。basicqos（prefetchcount1）消费消息channel。basicconsume（queuerpcqueue，onmessagecallbackonrequest）queue为队列名，onmessagecallback为回调函数，收到消息后，会调用回调函数print（〔x〕AwaitingRPCrequests）channel。startconsuming（）开始接收消息，进入阻塞状态，等待消息，直到收到消息为止，收到消息后，会调用onrequest函数客户端importpikaimportuuidclassFibonacciRpcClient（object）：definit（self）：指定用户名和密码，如果rabbitmq没有设置用户名和密码，可以不指定self。credentialspika。PlainCredentials（admin，admin）host指定rabbitmq服务器ip地址，credentials指定用户名和密码self。connectionpika。BlockingConnection（pika。ConnectionParameters（47。101。159。222，credentialsself。credentials））self。channelself。connection。channel（）声明一个随机队列，用来接收rpcserver返回的结果resultself。channel。queuedeclare（queue，exclusiveTrue）exclusiveTrue表示这个队列只能被当前连接使用，当连接关闭时，队列会被删除，exclusiveTrue是为了防止多个客户端同时使用一个队列获取随机队列的名称self。callbackqueueresult。method。queueself。channel。basicconsume（queueself。callbackqueue，onmessagecallbackself。onresponse，消费消息autoackTrue自动发送ack消息，告诉rabbitmq，消息已经被处理）defonresponse（self，ch，method，props，body）：ifself。corridprops。correlationid：self。responsebodydefcall（self，n）：self。responseNone生成一个随机的correlationid，用来标识消息，客户端和服务端都会用这个id来标识消息，客户端会将这个id传递给服务端，服务端会将这个id传递给客户端，客户端和服务端都会将这个id与自己的id进行比较，如果不一致，则丢弃这个消息self。corridstr（uuid。uuid4（））self。channel。basicpublish（exchange，routingkeyrpcqueue，将消息发送到rpcqueue队列propertiespika。BasicProperties（消息属性，用来标识消息replytoself。callbackqueue，将消息发送到客户端的回调函数，用来接收服务端返回的结果，服务端会将结果发送到这个队列correlationidself。corrid，将客户端的crrelationid发送给服务端），bodystr（n）将消息发送给服务端，服务端会将这个消息作为参数传递给fib函数）whileself。responseisNone：如果没有收到服务端返回的结果，则一直等待，直到收到结果，然后返回结果self。connection。processdataevents（）非阻塞版的startconsuming（），用来接收消息returnint（self。response）fibonaccirpcFibonacciRpcClient（）实例化客户端，用来发送消息，并接收服务端返回的结果，并返回结果，用来调用服务端的方法print（〔x〕Requestingfib（30））responsefibonaccirpc。call（10）调用call方法，发送消息，并接收服务端返回的结果，然后打印结果print（〔。〕Gotrresponse）
　　十二：python中的rpc框架python自带的：SimpleXMLRPCServer（数据包大，数据慢）HTTP通信第三方：ZeroRPC（底层使用ZeroMQ和MessagePack，速度快，响应时间短，并发高），grpc（谷歌推出支持夸语言）TCP通信十三：SimpleXMLRPCServer服务端fromxmlrpc。serverimportSimpleXMLRPCServerclassRPCServer（object）：definit（self）：初始化父类，python3中不需要，python2中需要，否则会报错super（RPCServer，self）。init（）print（self）self。senddata｛server：str（i）：iforiinrange（100）｝self。recvdataNonedefgetObj（self）：接收数据returnself。senddatadefsendObj（self，data）：发送数据print（senddata）self。recvdatadataprint（self。recvdata）创建一个服务器，监听本机的8000端口，并设置允许访问的ip地址，如果不设置，默认只能本机访问serverSimpleXMLRPCServer（（localhost，4242），allownoneTrue）注册内省函数，可以查看服务器提供的方法，不注册的话，客户端只能调用registerfunction注册的方法为什么要注册内省函数呢？因为客户端调用方法时，是通过方法名来调用的，如果不注册内省函数，客户端就不知道服务器提供了哪些方法server。registerintrospectionfunctions（）注册实例，可以调用实例的方法，不注册的话，客户端只能调用registerfunction注册的方法server。registerinstance（RPCServer（））开始监听请求，进入阻塞状态，等待请求，直到收到请求为止，收到请求后，会调用注册的方法server。serveforever（）客户端importtimefromxmlrpc。clientimportServerProxydefxmlrpcclient（）：print（xmlrpcclientstart）创建一个服务器代理，指定服务器的ip地址和端口cServerProxy（http：localhost：4242）调用服务器的方法data｛client：str（i）：iforiinrange（100）｝starttime。clock（）计时foriinrange（5）：重复调用50次ac。getObj（）调用服务器的方法print（a）foriinrange（5）：重复调用50次c。sendObj（data）调用服务器的方法print（xmlrpctotaltimes（time。clock（）start））ifnamemain：xmlrpcclient（）
　　十四：ZeroRPC实现rpc服务端importzerorpcclassRPCServer（object）：definit（self）：print（self）self。senddata｛server：str（i）：iforiinrange（100）｝self。recvdataNonedefgetObj（self）：print（getdata）returnself。senddatadefsendObj（self，data）：print（senddata）self。recvdatadataprint（self。recvdata）创建一个服务器，监听本机的8000端口，并设置允许访问的ip地址，如果不设置，默认只能本机访问szerorpc。Server（RPCServer（））绑定端口，并设置允许访问的ip地址，如果不设置，默认只能本机访问s。bind（tcp：0。0。0。0：4243）开始监听请求，进入阻塞状态，等待请求，直到收到请求为止，收到请求后，会调用注册的方法s。run（）客户端importzerorpcclassRPCServer（object）：definit（self）：print（self）self。senddata｛server：str（i）：iforiinrange（100）｝self。recvdataNonedefgetObj（self）：print（getdata）returnself。senddatadefsendObj（self，data）：print（senddata）self。recvdatadataprint（self。recvdata）创建一个服务器，监听本机的8000端口，并设置允许访问的ip地址，如果不设置，默认只能本机访问szerorpc。Server（RPCServer（））绑定端口，并设置允许访问的ip地址，如果不设置，默认只能本机访问s。bind（tcp：0。0。0。0：4243）开始监听请求，进入阻塞状态，等待请求，直到收到请求为止，收到请求后，会调用注册的方法s。run（）
　　十五：什么是RPC？1。RPC介绍？RPC是指远程过程调用，也就是说两台服务器，A和B，一个应用部署在A服务器上，想要调用B服务器上应用提供的函数或方法，由于不在一个内存空间，不能直接调用，需要通过网络来表达调用的语句和传达调用的数据。2。RPC是如何调用的？
　　1。要解决通讯的问题，主要是通过在客户端和服务器之间建立TCP连接，远程过程调用的所有交互的数据都在这个连接里传输。连接可以是按需连接，调用结束后就断掉，也可以是长连接，多个远程调用共享同一个连接。2。要解决寻址的问题，也就是说，A服务器上的应用怎么怎么告诉底层的RPC框架，如何连接到B服务器（如主机或IP地址）以及特定的端口，方法的名称是什么，这样才能完成调用。比如基于Wbe服务协议栈的RPC，就要提供一个endpointURl，或者是UDDI服务上查找。如果是RMI调用的话，还需要一个RMIRegistry来注册服务的地址。3。当A服务器上的应用发起远程过程调用时，方法的参数需要通过底层的网络协议如TCP传输到B服务器。由于网络协议是基于二进制的，内存中的参数的值要序列化成二进制形式，也就是序列化（Serialize）或编组（marshal），通过寻址和传输序列化的二进制发送给B服务器。4。B服务器收到请求后，需要对参数进行反序列化（序列化的逆操作），恢复为内存中的表达方式，然后找到对应的方法（寻址的一部分）进行本地调用，然后得到返回值。5。返回值还要发送回服务器A上的应用，也要经过序列化的方式发送，服务器A接收到后，再反序列化，恢复为内存中的表达方式，交给A服务器上的应用。3。为什么要使用RPC？就是无法在一个进程内，甚至一个计算机内通过本地调用的方式完成需求，比如不同的系统间的通讯，甚至不同的组织间的通讯。由于计算能力需要横向扩展，需要在多台机器组成的集群上部署应用。4。常见的RPC框架
　　功能
　　Hessian
　　Montan
　　rpcx
　　gRPC
　　Thrift
　　Dubbo
　　Dubbox
　　SpringCloud
　　开发语言
　　跨语言
　　Java
　　Go
　　跨语言
　　跨语言
　　Java
　　Java
　　Java
　　分布式（服务治理）
　　多序列化框架支持
　　hessian
　　（支持Hessian2、Json，可扩展）
　　只支持protobuf）
　　（thrift格式）
　　多种注册中心
　　管理中心
　　跨编程语言
　　（支持phpclient和Cserver）
　　支持REST
　　关注度
　　低
　　中
　　低
　　中
　　中
　　中
　　高
　　中
　　上手难度
　　低
　　低
　　中
　　中
　　中
　　低
　　低
　　中
　　运维成本
　　低
　　中
　　中
　　中
　　低
　　中
　　中
　　中
　　开源机构
　　Caucho
　　Weibo
　　Apache
　　Google
　　Apache
　　Alibaba
　　Dangdang
　　Apache5。实际的场景中的选择copySpringCloud：Spring全家桶，用起来很舒服，只有你想不到，没有它做不到。可惜因为发布的比较晚，国内还没出现比较成功的案例，大部分都是试水，不过毕竟有Spring作背书，还是比较看好。Dubbox：相对于Dubbo支持了REST，估计是很多公司选择Dubbox的一个重要原因之一，但如果使用Dubbo的RPC调用方式，服务间仍然会存在API强依赖，各有利弊，懂的取舍吧。Thrift：如果你比较高冷，完全可以基于Thrift自己搞一套抽象的自定义框架吧。Montan：可能因为出来的比较晚，目前除了新浪微博16年初发布的，Hessian：如果是初创公司或系统数量还没有超过5个，推荐选择这个，毕竟在开发速度、运维成本、上手难度等都是比较轻量、简单的，即使在以后迁移至SOA，也是无缝迁移。rpcxgRPC：在服务没有出现严重性能的问题下，或技术栈没有变更的情况下，可能一直不会引入，即使引入也只是小部分模块优化使用。
　　原文链接：https：www。cnblogs。comMyIronManp17015001。html