数据分析之道-基础知识字符串

Posted 2023-04-03 i阿极

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了数据分析之道-基础知识字符串相关的知识，希望对你有一定的参考价值。

文章目录

专栏导读
1、字符串介绍
2、访问字符串中的值
3、字符串拼接
4、转义字符
5、字符串运算符
6、字符串格式化
7、字符串内置函数

专栏导读

✍ 作者简介：i阿极，CSDN Python领域新星创作者，专注于分享python领域知识。

✍ 本文录入于《数据分析之道》，本专栏针对大学生、初级数据分析工程师精心打造，对python基础知识点逐一击破，不断学习，提升自我。
✍ 订阅后，可以阅读《数据分析之道》中全部文章内容，包含python基础语法、数据结构和文件操作，numpy科学计算，panda实现文件内容操作，matplotlib实现数据可视化。
✍ 还可以订阅进阶篇《数据分析之术》，其包含数据分析方法论、数据挖掘算法原理、业务分析实战。
✍ 其他专栏：《数据分析案例》，《机器学习案例》

1、字符串介绍

我们将介绍的第一种数据类型是字符串。字符串虽然看似简单，但能够以很多不同的方式使用。字符串就是一系列字符，是 Python 中最常用的数据类型。
在Python中，用引号括起的都是字符串，其中的引号可以是单引号，也可以是双引号。现在我们可以使用引号来创建字符串。创建字符串很简单，只要为变量分配一个值即可。例如：

str1 = “hello world”
str2 = “123456789”
str3 = “数据分析之道”

2、访问字符串中的值

访问字符串可以使用方括号[]来截取所需字符串

注意：索引值以 0 为开始值，-1 为从末尾的开始位置。

正向索引：

反向索引：

3、字符串拼接

字符串与字符串之间可以进行拼接，例如：

str = 'hello world'
print(str[:6] + '阿极')

4、转义字符

有一些字符因为在python中已经被定义为一些操作（比如单引号和双引号被用来引用字符串），而这些符号我们可能在字符串中需要使用到。为了能够使用这些特殊字符，可以用反斜杠 \\ 转义字符（同样地，反斜杠也可以用来转义反斜杠）。

常见的转义字符：

转义字符	描述
\\ (在行尾时)	续行符

print("line1 \\
 line2 \\
 line3")

转义字符	描述
\\ t	横向制表符

print("Hello \\t World!")

转义字符	描述
\\ n	换行符

print(' hello \\n world')

5、字符串运算符

操作符	描述
+	字符串连接
*	重复输出字符串
[]	通过索引获取字符串的字符
[:]	截取字符串中的一部分，遵循左闭右开原则，str[0:2] 是不包含第 3 个字符的。
in	成员运算符 - 如果字符串中包含给定的字符返回 True
not in	成员运算符 - 如果字符串中不包含给定的字符返回 True
%	格式字符串

实例：

a = "hello"
b = "world"
 
print("a + b 输出结果：", a + b)
print("a * 2 输出结果：", a * 2)
print("a[1] 输出结果：", a[1])
print("a[1:4] 输出结果：", a[1:4])
 
if( "h" in a) :
    print("h 在变量 a 中")
else :
    print("h 不在变量 a 中")
 
if( "m" not in a) :
    print("m 不在变量 a 中")
else :
    print("m 在变量 a 中")

6、字符串格式化

Python 支持格式化字符串的输出。尽管这样可能会用到非常复杂的表达式，但最基本的用法是将一个值插入到一个有字符串格式符 %s 的字符串中。

常见的符号：

符号	描述
%c	格式化字符及其ASCII码
%s	格式化字符串
%d	格式化整数
%f	格式化浮点数字，可指定小数点后的精度

实例：

name = "阿极"
age = 18
print ("我叫 %s 今年 %d 岁!" % (name,age))

7、字符串内置函数

常用的方法：

方法	描述
capitalize()	将字符串的第一个字符转换为大写
count(str, beg= 0,end=len(string))	返回 str 在 string 里面出现的次数，如果指定 beg 或者 end，则返回指定范围内 str 出现的次数
len(string)	返回字符串长度
lower()	转换字符串中所有大写字符为小写.
upper()	转换字符串中所有小写字符为大写.
join(seq)	以指定字符串作为分隔符，将 seq 中所有的元素(的字符串表示)合并为一个新的字符串
split(str=“”, num=string.count(str))	以 str 为分隔符截取字符串，如果 num 有指定值，则仅截取 num+1 个子字符串
strip([chars])	在字符串上执行 lstrip()和 rstrip()
swapcase()	将字符串中大写转换为小写，小写转换为大写

下一章，我们会讲列表基础知识

📢文章下方有交流学习区！一起学习进步！💪💪💪
📢创作不易，如果觉得文章不错，可以点赞👍收藏📁评论📒
📢你的支持和鼓励是我创作的动力❗❗❗

Java架构师成长之道之RabbitMQ开发与运维-基础篇(CSDN版)

Java架构师成长之道

消息中间件概述

消息是指在不同语言实现的应用间传递的数据，消息可以是文本字符串、JSON，也可以是复杂的内前对象。
消息中间件是指利用高效可靠的消息传递机制进行与平台无关的数据交流，并基于数据通讯来进行分布式系统集成。
通过提供消息传递和消息排队模型，使得在分布式环境下扩展进程间的通讯。

消息中间件一般有两种传递模式:点对点模式和发布/订阅模式。
点对点是基于队列的，消息生产者将消息发送到消息队列，消息消费者从队列中消费消息，队列的存在使得消息实现异步传输。
发布订阅模式定义了如果想一个内容节点发布和订阅消息，这个内容节点称为topic，主题可以理解称为消息传递的中介，消息发布者将消息发布到某个主题，而消息订阅者者从主题订阅消息，主题使得消息的订阅者和消息的发布者相互保持独立，不需要进行接触即可保持消息的传递，发布/订阅模式在消息的一对多广播时采用。

有了消息中间件之后，Application A和Application B可以使用消息中间件的API发送消息进行通讯。

应用通过消息中间件进行通讯

Application A可以通过消息中间件提供的API将消息发送给消息中间件服务器，然后消息中间件服务器将消息发送给 Application B，如果网络连接不可用，消息中间件会存储消息，直到连接变得可用，再将消息转发给应用程序B。
Application A发送消息时，Application B甚至可以处于不在运行的状态，消息中间件保留这份消息，直到Application B开始执行并消费消息，这样还防止了应用程序A因为等待应用程序B消费消息而出现阻塞。

主流消息中间件

目前主流的消息中间件有Kafka,RocketMQ,RabbitMQ

Kafka

LinkedIn开源的分布式发布-订阅消息系统，后来贡献给Apache,成为顶级开源项目
基于Pull模式处理消费消息
追求高吞吐量，一开始的目的就是用于日志收集和传输，适合大数据的互联网服务的数据收集业务
0.8版本开始支持复制
不支持事务，对于消息的重复、丢失、错误没有严格要求
高性能读写是基于操作系统底层的Page Cache实现，使用内存存储。

基于内存、高性能，节点之间相互复制
技术图片

RocketMQ

阿里开源的消息中间件，后来贡献给Apache，成为顶级开源项目
使用Java开发，具有高吞吐量、高可用性、适用大规模分布式系统的特点
RocketMQ思路源于Kakfa，对消息的可靠传输以及事务性做了优化
目前被阿里集团广泛应用于交易、充值、流计算、消息推送、日志流式处理、binglog分发等场景
商业版收费

RocketMQ集群拓扑图
技术图片

RabbitMQ

RabbitMQ是Erlang语言开发的开源消息队列系统，基于AMQP协议实现，用来通过普通协议在完全不同的应用之间共享数据
AMQP的主要特征是面向消息、队列、路由(包括点对点发布订阅)、可靠性、安全
AMQP协议更多用在企业系统内部，对数据一致性、稳定性、可靠性要求很高的场景，对性能和吞吐量的要求还在其次。
适用在金融交易场景

基于镜像队列实现高可用
技术图片

AMQP协议概述

AMQP协议的全称是：Advanced Message Queuing Protocol（高级消息队列协议）。目前AMQP协议的版本为 Version 1.0，这个协议标准在2014年通过了国际标准组织 (ISO) 和国际电工委员会 (IEC) 的投票，成为了新的 ISO 和 IEC 国际化标准。目前基于AMQP协议实现的产品包括:

Apache Qpid, an Apache project
Fedora Linux AMQP Infrastructure
IIT Software‘s SwiftMQ is a enterprise grade JMS messaging product with full support for AMQP 1.0
INETCO‘s AMQP protocol analyzer
JORAM: open reliable asynchronous messaging, 100% pure Java implementation of JMS
Kaazing‘s AMQP Web Client
Microsoft‘s Windows Azure Service Bus
JBoss A-MQ by Red Hat built from Qpid
StormMQ a cloud hosted messaging service based on AMQP
VMware Inc RabbitMQ; also supported by SpringSource
MQlight by IBM

关于AMPQ协议的完整介绍可以参考官方文档

AMQP协议中的元素包括：Message（消息体）、Producer（消息生产者）、Consumer（消息消费者）、Virtual Host（虚拟节点）、Exchange（交换机）、Queue（队列）等

由Producer（消息生产者）和Consumer（消息消费者）构成了AMQP的客户端，他们是发送消息和接收消息的主体。AMQP服务端称为Broker，一个Broker中一定包含完整的Virtual Host（虚拟主机）、 Exchange（交换机）、Queue（队列）定义。
一个Broker可以创建多个Virtual Host（虚拟主机），我们将讨论的Exchange和Queue都是虚拟机中的工作元素（还有User元素）。注意，如果AMQP是由多个Broker构成的集群提供服务，那么一个Virtual Host也可以由多个Broker共同构成。
Connection是由Producer（消息生产者）和Consumer（消息消费者）创建的连接，连接到Broker物理节点上。但是有了Connection后客户端还不能和服务器通信，在Connection之上客户端会创建Channel，连接到Virtual Host或者Queue上，这样客户端才能向Exchange发送消息或者从Queue接受消息。一个Connection上允许存在多个Channel，只有Channel中能够发送/接受消息。
Virtual Host-虚拟主机:虚拟地址，用于进行逻辑隔离，最上层的消息路由，一个Virtual Host可以有多个Exchange和Queue
同一个Virtual Host里面不能有同名的Exchange或Queue。
Exchange元素是AMQP协议中的交换机，Exchange可以绑定多个Queue也可以同时绑定其他Exchange。消息通过Exchange时，会按照Exchange中设置的Routing（路由）规则，将消息发送到符合的Queue或者Exchange中。

目前常用的Exchange Type有 Direct,Topic,Fanout三种类型

Direct Exchange

所有发送到Direct Exchange的消息被转发到Routing key中指定的Queue

Direct模式可以使用RabbitMQ自带的Exchange:default Exchange，所以不需要将Exchange进行任何绑定(bingding)操作，消息传递时，Routing key必须完全匹配才会被队列接收，否则该消息会被抛弃。

Topic Exchange
所有发送到Topic Exchange的消息被转发到所关心Routing Key中所指定Topic的Queue上。
Exchange将Routing key和某个Topic进行匹配，此时队列需要绑定一个Topic。
可以使用通配符进行模糊匹配
符号"#"匹配一个或者多个词
符号 "*"匹配一个词
例如log.#能够匹配到log.info.oa
log.*只会匹配到log.error

TopIic Exchange的路由规则
技术图片

Fanout Exchange
Facount Exchange不需要路由键，只需要简单的将队列绑定到交换机上,发送到该交换机的消息都会被转发到与该交换机绑定的队列上。Fanout交换机转发消息是最快的。
Binding-绑定:Exchange和Exchange、Queue之间的连接关系，绑定中可以包含routing key或者参数。
Queue:消息队列，实际存储消息数据，常用的属性有Durability:是否持久化，Durable:是，Transient:否,Auto delete:如果选yes,代表最后一个监听被移除之后，该Queue会自动删除。
Message:服务器和应用程序之间传送个的数据，本质上就是一段数据，由Properties和Payload(Body)组成，消息中常用的属性有 delivery mode,headers(自定义属性),content_type,content_encoding,priority,correlation_id,reply_to,expiration,message_id,timestamp,type,user_id,app_id,cluster_id

RabbitMQ概述

RabbitMQ是使用Erlang语言开发，Erlang语言广泛用于网络交换机领域,Erlang有着与原生Socket一样的延迟，这样使得RabbitMQ在Broker之间进行数据交互的性能非常优秀。
RabbitMQ凭借着以下特性在美团、滴滴、去哪儿、头条、艺龙等一线互联网大厂广泛使用

开源、性能优秀、稳定
提供可靠性消息投递模式，返回模式，
与Spring AMPQ完美的整合，API丰富。
集群模式丰富、支持表达式配置、HA模式、镜像队列模型。
保证数据不丢失的前提下做到高可靠性、可用性。

RabbitMQ整体架构模型
技术图片

AMPQ消息的流转流程
技术图片
1.在Producer（消息生产者）客户端建立了Channel后，就建立了到Broker上Virtual Host的连接。接下来Producer就可以向这个Virtual Host中的Exchange发送消息了。

2.Exchange（交换机）能够处理消息的前提是：它至少已经和某个Queue或者另外的Exchange形成了绑定关系，并设置好了到这些Queue和Excahnge的Routing（路由规则）。Excahnge中的Routing有三种模式，我们随后会讲到。在Exchange收到消息后，会根据设置的Routing（路由规则），将消息发送到符合要求的Queue或者Exchange中（路由规则还会和Message中的Routing Key属性配合使用）。

3.Queue收到消息后，可能会进行如下的处理：如果当前没有Consumer的Channel连接到这个Queue，那么Queue将会把这条消息进行存储直到有Channel被创建（AMQP协议的不同实现产品中，存储方式又不尽相同）；如果已经有Channel连接到这个Queue，那么消息将会按顺序被发送给这个Channel。

4.Consumer收到消息后，就可以进行消息的处理了。但是整个消息传递的过程还没有完成：视设置情况，Consumer在完成某一条消息的处理后，将需要手动的发送一条ACK消息给对应的Queue（当然您可以设置为自动发送，或者无需发送）。Queue在收到这条ACK信息后，才会认为这条消息处理成功，并将这条消息从Queue中移除；如果在对应的Channel断开后，Queue都没有这条消息的ACK信息，这条消息将会重新被发送给另外的Channel。当然，您还可以发送NACK信息，这样这条消息将会立即归队，并发送给另外的Channel。

RabbitMQ Server部署

CentOS系统基础配置

查看服务器版本
服务端操作系统基于CentOS7.3搭建，可以使用cat /etc/centos-release查看系统版本

[root@ittimeline ~]# cat /etc/centos-release
CentOS Linux release 7.3.1611 (Core)

配置hostname
然后使用vim /etc/hostname修改host配置,因为RabbitMQ Server日志的文件名会依赖hostname
vim编辑器的基本使用:
首先使用vim修改文件时进入的浏览模式，需要使用i键盘进入编辑模式
修改完成文件之后，需要使用esc退出编辑模式，然后使用:wq退出vim并保存修改的内容

[root@ittimeline ~]# vim  /etc/hostname

查看修改的内容

[root@ittimeline ~]# cat /etc/hostname                      
ittimeline.net

关闭防火墙
CentOS7的防火墙默认是开机启动的，这里为了方便使用RabbitMQ基于BS的管控台，需要关闭防火墙。

[root@ittimeline ~]# systemctl stop firewalld.service

还可以使用如下命令开机禁用防火墙

[root@ittimeline ~]# systemctl disable  firewalld.service

网卡开机启动

[root@ittimeline ~]# cd /etc/sysconfig/network-scripts/
 [root@ittimeline network-scripts]# cat ifcfg-eth0 
# Generated by parse-kickstart
DEVICE="eth0"
IPV6INIT="yes"
BOOTPROTO="dhcp"
UUID="c79432fa-c801-4f60-b4c1-a9cb34eeae2c"
ONBOOT="yes"
TYPE=Ethernet
DEFROUTE=yes
PEERDNS=yes
PEERROUTES=yes
IPV4_FAILURE_FATAL=no
IPV6_AUTOCONF=yes
IPV6_DEFROUTE=yes
IPV6_PEERDNS=yes
IPV6_PEERROUTES=yes
IPV6_FAILURE_FATAL=no
IPV6_PRIVACY=no
NAME="System eth0"

将属性ONBOOT的值改为yes即可，然后重启系统，就可以实现开机启动网络连接。

安装系统基础组件

在安装rabbitMQ之前使用yum install安装以下组件，否则rabbitMQ会安装失败。

[root@ittimeline ~]# yum install -y build-essential openssl openssl-devel unixODBC unixODBC-devel make gcc gcc-c++ kernel-devel m3 ncurse-devel tk tc xz

RabbitMQ 下载

在部署RabbitMQ之前需要下载erlang、socat、rabbitmq server三个组件

首先根据官网给出的Erlang和RabbitMQ版本对应关系
选择Erlang 22.0.X +RabbitMQ 3.7.17，版本确定以后，可以从官网下载erlang和rabbitmq安装包。

erlang针对CentOS7的rpm包
技术图片
abbitmq针对CentOS7的rpm包

下载erlang和rabbitmq只需要使用wget命令然后加上下载的地址即可
由于rabbitmq server依赖socat，还需要从阿里云镜像站上下载socat，注意它们的下载顺序和安装顺序一致

[root@ittimeline ~]# cd /home/liuguanglei/Downloads/
root@ittimeline Downloads]# wget --content-disposition https://packagecloud.io/rabbitmq/erlang/packages/el/7/erlang-22.0.7-1.el7.x86_64.rpm/download.rpm

[root@ittimeline Downloads]# wget https://mirrors.aliyun.com/centos/7.6.1810/os/x86_64/Packages/socat-1.7.3.2-2.el7.x86_64.rpm

[root@ittimeline Downloads]# wget --content-disposition https://packagecloud.io/rabbitmq/rabbitmq-server/packages/el/7/rabbitmq-server-3.7.17-1.el7.noarch.rpm/download.rpm

查看下载的软件版本

[root@ittimeline Downloads]# ls -al
total 29428
drwxr-xr-x.  2 liuguanglei liuguanglei     4096 Aug  6 11:50 .
drwx------. 16 liuguanglei liuguanglei     4096 Aug  6 10:19 ..
-rw-r--r--.  1 root        root        19384788 Jul 12 01:31 erlang-22.0.7-1.el7.x86_64.rpm
-rw-r--r--.  1 root        root        10438300 Jul 29 12:07 rabbitmq-server-3.7.17-1.el7.noarch.rpm
-rw-r--r--.  1 root        root          296632 Aug 11  2017 socat-1.7.3.2-2.el7.x86_64.rpm

RabbitMQ 安装

erlang,socat,rabbitmq server的安装只需要使用rpm -ivh命令安装即可。

[root@ittimeline Downloads]# rpm -ivh erlang-22.0.7-1.el7.x86_64.rpm 
[root@ittimeline Downloads]# rpm -ivh socat-1.7.3.2-2.el7.x86_64.rpm 
[root@ittimeline Downloads]# rpm -ivh rabbitmq-server-3.7.17-1.el7.noarch.rpm

RabbitMQ 配置

使用rpm安装方式安装的rabbitmq的配置默认路径是/usr/lib/rabbitmq/lib/rabbitmq_server-3.7.17/ebin/rabbit.app
这里暂时只是修改loopback_users,让guest用户能够使用ip登录，如果不修改的话，登录RabbitMQ 管控台会看到如下提示
技术图片

[root@ittimeline Downloads]# vim /usr/lib/rabbitmq/lib/rabbitmq_server-3.7.17/ebin/rabbit.app

修改之前
技术图片
修改之后

就是把loopback_users的["<<guest>>"]修改为[guest]即可

RabbitMQ 服务管理

启动rabbitmq-server
使用rabbitmq-server start & 启动rabbitmq服务

[root@ittimeline Downloads]# rabbitmq-server start &
[1] 25834
[root@ittimeline Downloads]# 
  ##  ##
  ##  ##      RabbitMQ 3.7.17. Copyright (C) 2007-2019 Pivotal Software, Inc.
  ##########  Licensed under the MPL.  See https://www.rabbitmq.com/
  ######  ##
  ##########  Logs: /var/log/rabbitmq/rabbit@ittimeline.log
                    /var/log/rabbitmq/rabbit@ittimeline_upgrade.log

              Starting broker...
 completed with 0 plugins.

RabbitMQ服务启动后会输出版本信息以及日志路径，其日志路径
Logs: /var/log/rabbitmq/rabbit@ittimeline.log
/var/log/rabbitmq/rabbit@ittimeline_upgrade.log

如果想要在终端上输入其他命令，回车即可，此时RabbitMQ服务进入后台运行，可以使用ps命令查看rabbitmq进程信息

Last login: Tue Aug  6 10:20:39 2019 from 172.16.237.111
[root@ittimeline ~]# ps -ef|grep rabbitmq
root     25834  4615  0 12:12 pts/1    00:00:00 /sbin/runuser -u rabbitmq -- /usr/lib/rabbitmq/bin/rabbitmq-server start
rabbitmq 25843 25834  0 12:12 pts/1    00:00:00 /bin/sh /usr/lib/rabbitmq/bin/rabbitmq-server start
rabbitmq 25940     1  0 12:12 ?        00:00:00 /usr/lib64/erlang/erts-10.4.4/bin/epmd -daemon
rabbitmq 25995 25843  0 12:12 pts/1    00:00:04 /usr/lib64/erlang/erts-10.4.4/bin/beam.smp -W w -A 64 -MBas ageffcbf -MHas ageffcbf -MBlmbcs 512 -MHlmbcs 512 -MMmcs 30 -P 1048576 -t 5000000 -stbt db -zdbbl 128000 -K true -B i -- -root /usr/lib64/erlang -progname erl -- -home /var/lib/rabbitmq -- -pa /usr/lib/rabbitmq/lib/rabbitmq_server-3.7.17/ebin  -noshell -noinput -s rabbit boot -sname rabbit@ittimeline -boot start_sasl -kernel inet_default_connect_options [nodelay,true] -sasl errlog_type error -sasl sasl_error_logger false -rabbit lager_log_root "/var/log/rabbitmq" -rabbit lager_default_file "/var/log/rabbitmq/rabbit@ittimeline.log" -rabbit lager_upgrade_file "/var/log/rabbitmq/rabbit@ittimeline_upgrade.log" -rabbit feature_flags_file "/var/lib/rabbitmq/mnesia/rabbit@ittimeline-feature_flags" -rabbit enabled_plugins_file "/etc/rabbitmq/enabled_plugins" -rabbit plugins_dir "/usr/lib/rabbitmq/plugins:/usr/lib/rabbitmq/lib/rabbitmq_server-3.7.17/plugins" -rabbit plugins_expand_dir "/var/lib/rabbitmq/mnesia/rabbit@ittimeline-plugins-expand" -os_mon start_cpu_sup false -os_mon start_disksup false -os_mon start_memsup false -mnesia dir "/var/lib/rabbitmq/mnesia/rabbit@ittimeline" -kernel inet_dist_listen_min 25672 -kernel inet_dist_listen_max 25672 start
rabbitmq 26097 25995  0 12:12 ?        00:00:00 erl_child_setup 1024
rabbitmq 26127 26097  0 12:12 ?        00:00:00 inet_gethost 4
rabbitmq 26128 26127  0 12:12 ?        00:00:00 inet_gethost 4
root     28310 28242  0 12:21 pts/0    00:00:00 grep --color=auto rabbitmq

使用lsof(list open files)查看进程进程打开的文件，进程打开的端口，打开文件的进程。。

[root@ittimeline ~]# lsof -i:5672
COMMAND    PID     USER   FD   TYPE DEVICE SIZE/OFF NODE NAME
beam.smp 25995 rabbitmq   77u  IPv6 148008      0t0  TCP *:amqp (LISTEN)

关闭rabbitmq server

[root@ittimeline Downloads]# rabbitmqctl stop
Stopping and halting node rabbit@ittimeline ...
Gracefully halting Erlang VM

启用 RabbitMQ Server Web管控台

使用rabbitmq-plugins enable rabbitmq_management启用RabbitMQ Server的Web管理界面

[root@ittimeline Downloads]# rabbitmq-plugins enable rabbitmq_management
Enabling plugins on node rabbit@ittimeline:
rabbitmq_management
The following plugins have been configured:
  rabbitmq_management
  rabbitmq_management_agent
  rabbitmq_web_dispatch
Applying plugin configuration to rabbit@ittimeline...
The following plugins have been enabled:
  rabbitmq_management
  rabbitmq_management_agent
  rabbitmq_web_dispatch

set 3 plugins.
Offline change; changes will take effect at broker restart.

然后就可以使用浏览器访问地址http://172.16.238.193:15672/
然后输入guest/guest后点击Login按钮登录RabbitMQ管控台
技术图片
RabbitMQ管控台预览页面

其中172.16.238.193是CentOS的ip地址，可以使用ifconfig命令查看

[root@ittimeline Downloads]# ifconfig
eth0: flags=4163<UP,BROADCAST,RUNNING,MULTICAST>  mtu 1500
        inet 172.16.238.193  netmask 255.255.252.0  broadcast 172.16.239.255
        inet6 fe80::21c:42ff:feb5:d22e  prefixlen 64  scopeid 0x20<link>
        ether 00:1c:42:b5:d2:2e  txqueuelen 1000  (Ethernet)
        RX packets 1110406  bytes 326972310 (311.8 MiB)
        RX errors 0  dropped 0  overruns 0  frame 0
        TX packets 39752  bytes 5885101 (5.6 MiB)
        TX errors 0  dropped 0 overruns 0  carrier 0  collisions 0

lo: flags=73<UP,LOOPBACK,RUNNING>  mtu 65536
        inet 127.0.0.1  netmask 255.0.0.0
        inet6 ::1  prefixlen 128  scopeid 0x10<host>
        loop  txqueuelen 1  (Local Loopback)
        RX packets 1616  bytes 105899 (103.4 KiB)
        RX errors 0  dropped 0  overruns 0  frame 0
        TX packets 1616  bytes 105899 (103.4 KiB)
        TX errors 0  dropped 0 overruns 0  carrier 0  collisions 0

virbr0: flags=4099<UP,BROADCAST,MULTICAST>  mtu 1500
        inet 192.168.122.1  netmask 255.255.255.0  broadcast 192.168.122.255
        ether 52:54:00:16:27:85  txqueuelen 1000  (Ethernet)
        RX packets 0  bytes 0 (0.0 B)
        RX errors 0  dropped 0  overruns 0  frame 0
        TX packets 0  bytes 0 (0.0 B)
        TX errors 0  dropped 0 overruns 0  carrier 0  collisions 0

其中eth0表示第一块网卡。
而15672是RabbitMQ管控台的访问端口，常用的端口还有客户端程序(Java,Python)通讯的5672端口和集群的25672端口。

基于Java Client 实现消息生产和消费

通用RabbitMQ连接管理实现

在rabbitmq-java-client-utils模块下的pom.xml文件中加入rabbitmq java-client以及slf4j和lombok的依赖

 <dependencies>
        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <scope>compile</scope>
        </dependency>
        
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
        </dependency>
        <dependency>
            <groupId>ch.qos.logback</groupId>
            <artifactId>logback-core</artifactId>
        </dependency>
        <dependency>
            <groupId>ch.qos.logback</groupId>
            <artifactId>logback-classic</artifactId>
        </dependency>
    </dependencies>

在rabbitmq-java-client-utils模块下的src/main/resources目录下定义rabbitmq.properties

同学可以根据自己的RabbitMQ Server配置作相应的修改

rabbitmq.host=172.16.238.74
rabbitmq.port=5672
rabbitmq.virtualhost=/
rabbitmq.username=guest
rabbitmq.password=guest

实现封装获取以及关闭连接的方法

package net.ittimeline.java.middleware.rabbitmq.java.utils;

import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.io.InputStream;
import java.util.Properties;
import java.util.concurrent.TimeoutException;

/**
 * Rabbit Message Queue 工具类
 * 封装获取连接和关闭连接方法
 *
 * @author liuguanglei 18601767221@163.com
 * @create 2019-08-06 19:53
 * @website www.ittimeline.net
 * @since JDK8u211
 * @see https://www.rabbitmq.com/java-client.html
 */
@Slf4j
public class RabbitMQUtils 


    /**
     * 基于rabbitmq java client构建客户端与rabbitmq server连接
     * @return
     */
    public static Connection getConnection()
        Connection connection=null;
        try(
                //读取src/main/resources目录下的rabbitmq.properties
                InputStream inputStream= RabbitMQUtils.class.getClassLoader().getResourceAsStream("rabbitmq.properties")
        )
            ConnectionFactory connectionFactory=new ConnectionFactory();
            Properties properties = new Properties();
            properties.load(inputStream);
            connectionFactory.setHost(properties.getProperty("rabbitmq.host"));
            connectionFactory.setPort(Integer.valueOf(properties.getProperty("rabbitmq.port")));
            connectionFactory.setAutomaticRecoveryEnabled(true);
            connectionFactory.setNetworkRecoveryInterval(3000);
            connectionFactory.setVirtualHost(properties.getProperty("rabbitmq.virtualhost"));
            connectionFactory.setUsername(properties.getProperty("rabbitmq.username"));
            connectionFactory.setPassword(properties.getProperty("rabbitmq.password"));
            connection=connectionFactory.newConnection();
        catch (IOException | TimeoutException e)

            log.error("获取RabbitMQ Connection失败",e);

        

        return connection;
    


    /**
     * 关闭Rabbit Server 连接
     * @param connection
     */
    public static void close(Connection connection)
        if(connection!=null)
            try 
                if (connection.isOpen())
                    connection.close();
                
               
             catch (IOException e) 

                log.error("关闭RabbitMQ连接失败",e);

基于Default Exchange 实现消息生产和消费

Default Exchange 默认交换隐式绑定到每个队列，路由密钥等于队列名称。无法显式绑定到默认交换或从绑定取消绑定。它也无法删除。

实现基于RabbitMQ java-client的消息生产端

package net.ittimeline.java.mq.middleware.rabbitmq.java.client.producer;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.MessageProperties;
import lombok.extern.slf4j.Slf4j;
import net.ittimeline.java.middleware.rabbitmq.java.utils.RabbitMQUtils;

import java.io.IOException;

/**
 * 发送消息
 * 指定exchange和routing key以及消息的内容
 *
 * @author liuguanglei 18601767221@163.com
 * @create 2019-08-06 17:50
 * @website www.ittimeline.net
 * @since JDK8u211
 */
@Slf4j
public class ProducerMessage 


    public static void main(String[] args) 
        Connection connection = RabbitMQUtils.getConnection();

        try 


            /**
             * 交换器的名称
             * 指明消息需要发送到哪个交换器中
             * 如果设置为空字符串，则消息会发送到RabbitMQ默认的交换器中 AMQP Exchange
             * 默认交换隐式绑定到每个队列，路由密钥等于队列名称。无法显式绑定到默认交换或从绑定取消绑定。它也无法删除。
             */
            String exchangeName = "";
            /**
             * 路由键
             * 交换器根据路由键将消息存储到相应的队列中
             * routingKey的值和QueueName相同
             */
            String routingKey = "ittimeline";

            String message = "Hello,RabbitMQ";
            /**
             * 消息体,真正需要发送消息
             */
            byte[] messageBody = message.getBytes();
            Channel channel = connection.createChannel();
            //声明队列 消息最终由Exchange路由到队列上
            String queueName = "ittimeline";
            channel.queueDeclare(queueName, true, false, false, null);

            channel.basicPublish(exchangeName, routingKey, true, MessageProperties.PERSISTENT_TEXT_PLAIN, messageBody);
            log.info("发送RabbitMQ消息成功 ，消息的内容是", message);

         catch (Exception e) 
            log.error("ProducerMessage发送RabbitMQ消息失败", e);
         finally 
            RabbitMQUtils.close(connection);

实现基于RabbitMQ java-client端实现的消息消费端

package net.ittimeline.java.middleware.rabbitmq.java.client.consumer;

import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;
import net.ittimeline.java.middleware.rabbitmq.java.utils.RabbitMQUtils;

import java.io.IOException;

/**
 * 消费消息
 *
 * @author liuguanglei 18601767221@163.com
 * @create 2019-08-06 17:50
 * @website www.ittimeline.net
 * @since JDK8u211
 */
@Slf4j
public class ConsumerMessage 

    public static void main(String[] args) 


        Connection connection= RabbitMQUtils.getConnection();
        try 
            Channel channel=connection.createChannel();
            String queueName="ittimeline";
//            //声明(创建)一个队列  启动时先启动Consumer,因为必须要先有队列,生产者才能将消息通过Exchange发送到队列上
            channel.queueDeclare(queueName,true,false,false,null);
            //创建消费者
            channel.basicConsume(queueName,new DefaultConsumer(channel)
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException 
                    String routingKey=envelope.getRoutingKey();
                    String exchange=envelope.getExchange();
                    log.info("收到RabbitMQ消息,消息的exchange:，消息的routingKey:，消息的内容:",exchange,routingKey,new String(body));

                
            );
         catch (Exception e) 
           log.error("ConsumerMessage接收RabbitMQ消息失败，原因是",e);
        
        finally
           // RabbitMQUtils.close(connection);

基于 Direct Exchange实现消息生产和消费

基于Direct Exchange的消息生产和消费,Producer和Consumer的routing key 必须完全一致

基于Direct Exchange 实现消息生产者

package net.ittimeline.java.mq.middleware.rabbitmq.java.client.producer.exchange;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import lombok.extern.slf4j.Slf4j;
import net.ittimeline.java.middleware.rabbitmq.java.utils.RabbitMQUtils;

import java.io.IOException;

/**
 * 基于Direct Exchange 生产消息
 * Producer和Consumer的routing key必须完全一致
 *
 * @author liuguanglei 18601767221@163.com
 * @create 2019-08-07 14:00
 * @website www.ittimeline.net
 * @since JDK8u211
 */

@Slf4j
public class DirectExchangeProducerMessage 


    public static void main(String[] args) 

        try 
            Connection connection = RabbitMQUtils.getConnection();
            Channel channel = connection.createChannel();

//            //设置exchange和队列属性
//            String exchangeName = "test_direct_exchange";
//            String routingKey = "test.direct";
            String exchangeName="test_direct_exchange";
            String exchangeType="direct";
            String queueName="test_direct_queue";
            String routingKey="test.direct";

            //声明exchange
            channel.exchangeDeclare(exchangeName,exchangeType,false,false,false,null);
            //声明队列
            channel.queueDeclare(queueName,true,false,false,null);
            //根据routing key 绑定exchange和队列
            channel.queueBind(queueName,exchangeName,routingKey);

            String message = "Hello,RabbitMQ With Direct Exchange";

            channel.basicPublish(exchangeName, routingKey, null, message.getBytes());
            log.info("发送RabbitMQ消息成功,消息的内容是", message);

         catch (IOException e) 
            log.error("DirectExchangeProducerMessage发送RabbitMQ消息失败", e);

基于Direct Exchange 实现消息消费者

package net.ittimeline.java.middleware.rabbitmq.java.client.consumer.exchange;

import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;
import net.ittimeline.java.middleware.rabbitmq.java.utils.RabbitMQUtils;

import java.io.IOException;

/**
 * 基于direct exchange消费消息
 *Producer和Consumer的routing key必须完全一致
 * @author liuguanglei 18601767221@163.com
 * @create 2019-08-07 14:05
 * @website www.ittimeline.net
 * @since JDK8u211
 */
@Slf4j
public class DirectExchangeConsumerMessage 


    public static void main(String[] args) 

        Connection connection= RabbitMQUtils.getConnection();

        try 

            Channel channel=connection.createChannel();

            boolean autoAck=false;

            String exchangeName="test_direct_exchange";
            String exchangeType="direct";
            String queueName="test_direct_queue";
            String routingKey="test.direct";

            //声明exchange
            channel.exchangeDeclare(exchangeName,exchangeType,false,false,false,null);
            //声明队列
            channel.queueDeclare(queueName,true,false,false,null);
            //根据routing key 绑定exchange和队列
            channel.queueBind(queueName,exchangeName,routingKey);

            channel.basicConsume(queueName,autoAck,new DefaultConsumer(channel)
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException 

                    log.info("收到RabbitMQ消息成功,内容是 ",new String(body));
                
            );

         catch (IOException e) 
            log.error("DirectExchangeConsumerMessage接收RabbitMQ消息失败",e);

基于Topic Exchange实现消息生产和消费

消息消费端配置account.#,消息生产端配置account.get或者account.list.info都是可以收到消息
消息消费端配置account.*,消息生产端端配置account.get可以收到，但是account.list.info就收不到了，因为account.*只能匹配一个词

因为队列的声明只在TopicExchangeConsumerMessage类中定义，因此运行程序时需要先启动TopicExchangeConsumerMessage，再启动TopicExchangeProducerMessager。
否则第一次发送消息时因为队列还没有创建，导致Exchange无法将消息根据routing key路由到指定的队列中。

基于Topic Exchange实现消息生产者

package net.ittimeline.java.mq.middleware.rabbitmq.java.client.producer.exchange;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import lombok.extern.slf4j.Slf4j;
import net.ittimeline.java.middleware.rabbitmq.java.utils.RabbitMQUtils;

import java.io.IOException;

/**
 * 基于topic Exchange生产消息
 * 消息消费端配置account.#,消息生产端配置account.get或者account.list.info都是可以收到消息
 * 消息消费端配置account.*,消息生产端端配置account.get可以收到，但是account.list.info就收不到了，因为account.*只能匹配一个词
 *
 * @author liuguanglei 18601767221@163.com
 * @create 2019-08-07 14:48
 * @website www.ittimeline.net
 * @since JDK8u211
 */
@Slf4j
public class TopicExchangeProducerMessage 

    public static void main(String[] args) 

        try 

            Connection connection = RabbitMQUtils.getConnection();

            Channel channel = connection.createChannel();
            String exchangeName = "test_topic_exchange";

            final String info = "发送RabbitMQ消息成功,消息的内容是,routingKey是";

            String routingKey = "account.get";
            String message = "Hello RabbitMQ With Topic Exchange";


            channel.basicPublish(exchangeName, routingKey, null, message.getBytes());
            log.info(info, message, routingKey);

            routingKey = "account.save";
            channel.basicPublish(exchangeName, routingKey, null, message.getBytes());
            log.info(info, message, routingKey);

            routingKey = "account.update";
            channel.basicPublish(exchangeName, routingKey, null, message.getBytes());
            log.info(info, message, routingKey);

            routingKey = "account.delete";
            channel.basicPublish(exchangeName, routingKey, null, message.getBytes());
            log.info(info, message, routingKey);


            routingKey = "account.list.info";




            channel.basicPublish(exchangeName, routingKey, null, message.getBytes());
            log.info(info, message, routingKey);

         catch (Exception e) 
            log.error("TopicExchangeProducerMessage发送失败，原因是", e);

基于 Topic Exchange 实现的消息消费者

package net.ittimeline.java.middleware.rabbitmq.java.client.consumer.exchange;

import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;
import net.ittimeline.java.middleware.rabbitmq.java.utils.RabbitMQUtils;

import java.io.IOException;

/**
 * 基于topic exchange 消费消息
 * 消息消费端配置account.#,消息生产端配置account.get或者account.list.info都是可以收到消息
 * 消息消费端配置account.*,消息生产端端配置account.get可以收到，但是account.list.info就收不到了，因为account.*只能匹配一个词
 *
 * @author liuguanglei 18601767221@163.com
 * @create 2019-08-07 15:03
 * @website www.ittimeline.net
 * @since JDK8u211
 */
@Slf4j
public class TopicExchangeConsumerMessage 

    public static void main(String[] args) 

        try 
            Connection connection = RabbitMQUtils.getConnection();
            Channel channel = connection.createChannel();

            String exchangeName = "test_topic_exchange";
            String exchangeType = "topic";

            String queueName = "test_topic_queue";
            String routingKey = "account.#";
            //如果routingKey换成account.* 那么应该是收不到生产消息的routingKey为 account.list.info，因为*只能匹配一个词
            // String routingKey="account.*";

            //声明exchange
            channel.exchangeDeclare(exchangeName, exchangeType, true, false, false, null);
            //声明queue
            channel.queueDeclare(queueName, true, false, false, null);
            //根据routingKey绑定exchange和queue
            channel.queueBind(queueName, exchangeName, routingKey);

            channel.basicConsume(queueName, true, new DefaultConsumer(channel) 
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException 

                    log.info("收到消息的内容是", new String(body));
                
            );


         catch (Exception e) 
            log.error("TopicExchangeConsumerMessage接收消息失败，失败的原因是", e);

基于Fanout Exchange实现的消息生产和消费

Fanout Exchange不处理路由键，只需要简单的将队列绑定到交换机上，发送到交换机的消息都会转发到与该交换机绑定的所有队列上，Fanout交换机转发的消息是最快的

基于Fanout Exchange 实现的消息生产者

package net.ittimeline.java.mq.middleware.rabbitmq.java.client.producer.exchange;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import lombok.extern.slf4j.Slf4j;
import net.ittimeline.java.middleware.rabbitmq.java.utils.RabbitMQUtils;

import java.io.IOException;

/**
 * 基于fanout exchange的消息生产端
 * @author liuguanglei 18601767221@163.com
 * @create 2019-08-07 16:53
 * @website www.ittimeline.net
 * @since JDK8u211
 */
@Slf4j
public class FanoutExchangeProducerMessage 

    public static void main(String[] args) 

        Connection connection= RabbitMQUtils.getConnection();
        try 
            Channel channel=connection.createChannel();

            String exchangeName="test_fanout_exchange";
            //不设置路由键
            //String routingKey="";
            //或者设置任意的路由键
            //消息消费方都可以收到
            String routingKey="any";
            String message="Hello RabbitMQ With Fanout Exchange";
            channel.basicPublish(exchangeName,routingKey,null,message.getBytes());
            log.info("RabbitMQ消息发送成功，发送的内容是",message);

         catch (Exception e) 
           log.error("FanoutExchangeProducerMessage发送RabbitMQ消息失败,原因是",e);

基于Fanout Exchange 实现的消息消费者

package net.ittimeline.java.middleware.rabbitmq.java.client.consumer.exchange;

import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;
import net.ittimeline.java.middleware.rabbitmq.java.utils.RabbitMQUtils;

import java.io.IOException;

/**
 * 基于fanout exchange的消息消费端
 * @author liuguanglei 18601767221@163.com
 * @create 2019-08-07 16:54
 * @website www.ittimeline.net
 * @since JDK8u211
 */
@Slf4j
public class FanoutExchangeConsumerMessage 

    public static void main(String[] args) 
        try 
            Connection connection= RabbitMQUtils.getConnection();

            Channel channel=connection.createChannel();
            String exchangeName="test_fanout_exchange";
            String exchangeType="fanout";
            //不设置路由键
            String routingKey="";
            String queueName="test_fanout_queue";

            channel.exchangeDeclare(exchangeName,exchangeType,false,false,null);
            channel.queueDeclare(queueName,false,false,false,null);
            channel.queueBind(queueName,exchangeName,routingKey);

            channel.basicConsume(queueName,true,new DefaultConsumer(channel)
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException 
                    log.info("收到消息的内容是",new String(body));
                
            );


         catch (Exception e) 
            log.error("FanoutExchangeConsumerMessage接收RabbitMQ消息失败，原因是",e);

自定义附加属性消息生产和消费者

消息生产者-设置消息附加属性

package net.ittimeline.java.mq.middleware.rabbitmq.java.client.producer.properties;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import lombok.extern.slf4j.Slf4j;
import net.ittimeline.java.middleware.rabbitmq.java.utils.RabbitMQUtils;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

/**
 * 发送附加属性的消息
 *
 * @author liuguanglei 18601767221@163.com
 * @create 2019-08-06 17:50
 * @website www.ittimeline.net
 * @since JDK8u211
 */
@Slf4j
public class ProducerPropertiesMessage 


    public static void main(String[] args) 

        Connection connection = RabbitMQUtils.getConnection();
        /**
         * 交换器的名称
         * 指明消息需要发送到哪个交换器中
         * 如果设置为空字符串，则消息会发送到RabbitMQ默认的交换器中 AMQP Exchange
         * 默认交换隐式绑定到每个队列，路由密钥等于队列名称。无法显式绑定到默认交换或从绑定取消绑定。它也无法删除。
         */
        String exchangeName = "";
        /**
         * 路由键
         * 交换器根据路由键将消息存储到相应的队列中
         * routingKey的值和QueueName相同
         */
        String routingKey = "customPropertiesMessage";


        String message = "Hello,RabbitMQ";
        /**
         * 消息体,真正需要发送消息
         */
        byte[] messageBody = message.getBytes();

        //自定义属性
        Map<String, Object> headers = new HashMap<>(2);
        headers.put("username", "tony");
        headers.put("password", "666666");

        //配置消息的属性信息
        //重点关注消息的有效期，当前为10秒钟
        AMQP.BasicProperties messgeProperties = new AMQP.BasicProperties().builder().deliveryMode(2).contentEncoding("utf-8").expiration("10000").headers(headers).build();

        try (Channel channel = connection.createChannel()) 
            channel.basicPublish(exchangeName, routingKey, true, messgeProperties, messageBody);
            log.info("发送RabbitMQ消息成功 ，消息的内容是", message);

         catch (Exception e) 

            log.error("ProducerPropertiesMessage发送RabbitMQ消息失败", e);
         finally 
             RabbitMQUtils.close(connection);

消息消费者-获取消息附加属性

package net.ittimeline.java.middleware.rabbitmq.java.client.consumer.properties;

import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;
import net.ittimeline.java.middleware.rabbitmq.java.utils.RabbitMQUtils;

import java.io.IOException;
import java.util.Map;

/**
 * 消费自定义属性的消息
 *
 * @author liuguanglei 18601767221@163.com
 * @create 2019-08-06 17:50
 * @website www.ittimeline.net
 * @since JDK8u211
 */
@Slf4j
public class ConsumerPropertiesMessage 

    public static void main(String[] args) 


        try 
            Connection connection= RabbitMQUtils.getConnection();
            Channel channel=connection.createChannel();
            String queueName="customPropertiesMessage";
            //声明(创建)一个队列  启动时先启动Consumer,因为必须要先有队列,生产者才能将消息通过Exchange发送到队列上
            channel.queueDeclare(queueName,true,false,false,null);
            //创建消费者
            channel.basicConsume(queueName,new DefaultConsumer(channel)
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException 
                    String routingKey=envelope.getRoutingKey();
                    String expiration=properties.getExpiration();
                    String message=new String(body);
                    log.info("ConsumerPropertiesMessage接收RabbitMQ消息，消息的routingKey:,消息的过期属性:,消息的内容",routingKey,expiration,message);
                    //获取自定义属性
                    Map<String,Object> headers =properties.getHeaders();

                    if(null!=headers)
                        log.info("username ",headers.get("username"));
                        log.info("password ",headers.get("password"));
                    

                
            );
         catch (IOException e) 
           log.error("ConsumerPropertiesMessage接收RabbitMQ消息失败，原因是",e);
        
        finally
           // RabbitMQUtils.close(connection);

以上是关于数据分析之道-基础知识字符串的主要内容，如果未能解决你的问题，请参考以下文章