第一种，把数据库中存放的相关数据，全部读入PHP/Java/C#代码/内存，再由代码对其进行分页操作（速度慢，简易性高）。

第二种，直接在数据库中对相关数据进行分页操作，再把分页后的数据输出给代码程序（速度中，简易性中）。

第三种，先把数据库中的相关数据全部读入“缓存”或第三方工具，再由代码程序对“缓存”或第三方工具中的数据进行读取+分页操作（速度快，简易性差）。

本文下面重点阐述上述【第二种】方案在SQL Server上的使用（其它种类数据库由于Sql语句略有差异，所以需要调整，但方案也类似）

1、ROW_NUMBER() OVER()方式（SQL2012以下推荐使用）

示例：

SELECT * FROM (SELECT ROW_NUMBER() OVER(ORDER BY menuId) AS RowId,* FROM sys_menu ) AS r WHERE RowId BETWEEN 1 AND 10

用子查询新增一列行号（ROW_NUMBER）RowId查询，比较高效的查询方式，只有在SQL Server2005或更高版本才支持。

BETWEEN 1 AND 10 是指查询第1到第10条数据（闭区间），在这里面需要注意的是OVER的括号里面可以写多个排序字段。

通用用法

--pageIndex 表示指定页

--pageSize 表示每页显示的条数

SELECT * FROM (SELECT ROW_NUMBER() OVER(ORDER BY 排序字段) AS RowId,* FROM 表名 ) AS r WHERE RowId BETWEEN ((pageIndex-1)*pageSize + 1) AND (pageIndex * PageSize)

2、offset fetch next方式（SQL2012及以上的版本才支持：推荐使用 ）

示例：

--offset fetch next方式查询，最高效的查询方式，只有在SQL Server2012或更高版本才支持

SELECT * FROM sys_menu ORDER BY menuId offset 0 ROWS FETCH NEXT 10 ROWS ONLY

offset 是跳过多少行，

next是取接下来的多少行，

句式 offset...rows fetch nect ..rows only ，注意rows和末尾的only 不要写漏掉了，并且这种方式必须要接着Order by XX 使用，不然会报错。

通用用法

--pageIndex 表示指定页

--pageSize 表示每页显示的条数

SELECT * FROM 表名 ORDER BY 排序字段 offset ((pageIndex - 1) * pageSize) ROWS FETCH NEXT pageSize ROWS ONLY

3、top not in方式 （不推荐）

示例：

--查询第11-20条记录

SELECT TOP 10 menuId, *F

ROM sys_menu

WHERE menuId NOT IN (SELECT TOP 10 menuId FROM sys_menu)

这条语句的原理是先查询1-10条记录的ID，然后再查询ID不属于这1-10条记录的ID，并且只需要10条记录，因为每页大小就是10，

这就是获取到的第11-20条记录，这是非常简单的一种写法。

另外IN语句与NOT IN语句类似，这是NOT IN的写法，但是这种写法数据量大的话效率太低。

通用用法

--pageIndex 表示指定页

--pageSize 表示每页显示的条数

SELECT TOP pageSize menuId, *FROM sys_menu WHERE menuId NOT IN (SELECT TOP ((pageSize-1)*pageIndex) menuId FROM sys_menu)

4、通过升序与降序方式进行查询分页（不推荐）

示例：

--查询第11-20条记录

SELECT * FROM( SELECT TOP 10 * FROM( SELECT TOP 20 * FROM sys_menu ORDER BY menuId ASC) AS TEMP1 ORDER BY menuId DESC) AS TEMP2 ORDER BY menuId ASC

这条语句首先查询前20条记录，然后在倒序查询前10条记录（即倒数10条记录），

这个时候就已经获取到了11-20条记录，但是他们的顺序是倒序，所以最后又进行升序排序。

通用方法

--pageIndex 表示指定页

--pageSize 表示每页显示的条数

SELECT * FROM( SELECT TOP pageSize * FROM( SELECT TOP ((pageIndex - 1) * pageSize +(pageSize*2)) * FROM sys_menu

ORDER BY menuId ASC) AS TEMP1 ORDER BY menuId DESC) AS TEMP2 ORDER BY menuId ASC

5、采用MAX(ID)或者MIN(ID)函数（不推荐）

示例：

--查询第11-20条记录

SELECT TOP 10 * FROM sys_menu WHERE menuId> (SELECT MAX(menuId) FROM(SELECT TOP 10 menuId FROM sys_menu ORDER BY menuId) AS TEMP1) --（第10条的id）

这个理解起来也简单，先把第10条记录的id找出来（当然这里面是直接使用MAX()进行查找，MIN()函数的用法也是类似的），

然后再对比取比第10条记录的id大的前10条记录即为我们需要的结果。

这里要注意开始时的边界值调整。

通用用法

--pageIndex 表示指定页

--pageSize 表示每页显示的条数

SELECT TOP pageSize * FROM sys_menu WHERE menuId> (SELECT MAX(menuId) FROM(SELECT TOP ((PageIndex-1)*PageSize) menuId FROM sys_menu ORDER BY menuId) AS TEMP1) --（第10条的id）

上述1~5方案，再配合存储过程，你就能制造出适合你自己的“分页”轮子，日后反复使用。

但它们有一定自身局限性：方案1、2、5都需要依赖一个排序Id（这个Id要么是个排序列，要么是个主键）。方案3、4则效率太低，完全不推荐。

7、不依赖排序/排序Id的终极方案

此方案在DeveloperSharp框架中有提供（基于.Net/.Net Core/.Net Framework），方案被广东省的多个公司/项目采用，得到了实战验证+稳定性。

【第一步】：从NuGet引用DeveloperSharp包。

【第二步】：创建一个用来与数据库进行通信的“数据源类”（文本示例为：TestData.cs），内容如下：

using DeveloperSharp.Structure.Model;

using DeveloperSharp.Framework.QueryEngine;

namespace YZZ{

[DataSource(DatabaseType.SQLServer, "Server=localhost;Database=Test;Uid=sa;Pwd=123")]

public class TestData : DeveloperSharp.Structure.Model.DataLayer { //类中没有任何代码 }}

说 明 ：“数据源类”（文本示例为：TestData.cs）必 须 继 承 自 DeveloperSharp.Structure.Model.DataLayer 类 ， 并 且 在 其 上 设 置DataSource属 性 的 初 始 化 值 为“数据库类型”及其“链接字符串”。

【第三步】：添加通过“数据源类”（TestData）调用其PagePartition方法进行数据分页的代码。注 意：核心代码就一行而已！！

代码如下：

using DeveloperSharp.Extension;//Table扩展所在的命名空间----------------------------- 
class Program { 
               static void Main(string[] args) { TestData td = new TestData(); //分页 var pp = td.PagePartition("select top 5000 * from t_Order where Id>10 order by Id desc", 20, 162); List<Product> Products = pp.Table.ToList<Product>(); foreach (var P in Products) { Console.WriteLine(P.Name); } Console.ReadLine(); }
 }

Product类代码如下：

public class Product { public string Id { get; set; } public string Name { get; set; } public int Quantity { get; set; } }

此处的PagePartition方法有两个重载方法，其详细功能说明如下：

PagePartition声明：public PagePiece PagePartition(string RecordSet, string Id, int PageSize, int PageIndex)用途：分页功能(有主键)参数：（1）string RecordSet --需要分页的记录集，可以是表、视图、或者SQL语句（2）string Id --主键（3）int PageSize --页面大小（4）int PageIndex --当前页码返回：PagePiece --页片实体 PagePartition声明：public PagePiece PagePartition(string RecordSet, int PageSize, int PageIndex)用途：分页功能(无主键)参数：（1）string RecordSet -- 需要分页的记录集，可以是表、视图、或者SQL语句 （2）int PageSize --页面大小（3）int PageIndex --当前页码返回：PagePiece --页片实体

注意：

（1）     当你需要分页的数据表有“主键”字段时，使用“分页功能(有主键)”。反之，使用“分页功能(无主键)”。

（2）     RecordSet是你需要分页的“数据总集”的SQL语句。该SQL语句的形式丰富多样，可以带条件、排序、甚至还能是多表的联合查询、等。

（3）     此方法符合最开始的【第二种】方案，是在SQL Server内部进行的分页操作。而且可以不依赖于排序/排序Id。

分别是NotifyIcon控件和ContextMenuStrip控件

2、NotifyIcon 设置ico图标

3、 在控件NotifyIcon中绑定上ContextMenuStrip控件

当运行程序时，就可以看见图标显示在右下角了

以下代码根据实际需要添加

      
 private void Form1_Load(object sender, EventArgs e)
        {

            ToolStripMenuItem itm2 = new ToolStripMenuItem();
            itm2.Name = "showthis";
            itm2.Text = "显示窗体";
            itm2.Click += ShowthisDialog;
            myMenu.Items.Add(itm2);

｝

 private void notifyIcon1_MouseDoubleClick(object sender, MouseEventArgs e)
        {
            if (e.Button == MouseButtons.Left)
            {
                if (this.WindowState == FormWindowState.Normal)
                {
                    this.WindowState = FormWindowState.Minimized;
                    this.Hide();

                }
                else if (this.WindowState == FormWindowState.Minimized)
                {
                    this.Show();
                    this.WindowState = FormWindowState.Normal;
                    this.Activate();
                }
            }
            else if (e.Button == MouseButtons.Right)
            {

                myMenu.Show();

            }
        }

        private void exitPram_Click(object sender, EventArgs e)
        {
            DialogResult result = MessageBox.Show("您确认要关闭吗？关闭后将无法接收和处理数据", "警告", MessageBoxButtons.YesNo, MessageBoxIcon.Warning);
            if (result == DialogResult.Yes)
            {
                Application.Exit();
                // this.notifyIcon1.Dispose();
            }
        }

#region 显示隐藏窗体
        private void ShowthisDialog(object sender, EventArgs e)
        {
            this.Show();

            if (myMenu.Items.ContainsKey("showthis"))
            {
                myMenu.Items.RemoveByKey("showthis");
            }
            ToolStripMenuItem itm1 = new ToolStripMenuItem();
            itm1.Name = "hidethis";
            itm1.Text = "隐藏窗体";
            itm1.Click += HidethisDialog;
            myMenu.Items.Add(itm1);
            //this.Hide(); 
        }
        private void HidethisDialog(object sender, EventArgs e)
        {
            this.Hide();
            if (myMenu.Items.ContainsKey("hidethis"))
            {
                myMenu.Items.RemoveByKey("hidethis");
            }
            ToolStripMenuItem itm2 = new ToolStripMenuItem();
            itm2.Name = "showthis";
            itm2.Text = "显示窗体";
            itm2.Click += ShowthisDialog;
            myMenu.Items.Add(itm2);
        }
        private void startFrm_Shown(object sender, EventArgs e)
        {
            // textBox1.Text += "Form1_Shown" + System.Environment.NewLine;
            this.Hide();
        }
        #endregion

简单的讲就是：被遍历的集合在遍历过程中，被其他线程或者地方改变

public void ForeachDic()     
{     
     Dictionary<String, Int32> dic = new Dictionary<String, Int32>();   
     dic.Add("1", 10);     
     dic.Add("2", 20);     
     dic.Add("3", 30);     
     foreach (KeyValuePair<String, Int32> kvp in dic)     
     {     
         Console.WriteLine(String.Format("Key:{0}; Value:{1}", kvp.Key, kvp.Value));     
         dic[kvp.Key] = 100;//此操作会报错:集合已修改;可能无法执行枚举操作。    
     }     
}
  

解决办法
1、通常情况下，不要在枚举集合的操作，比如Foreach中去执行修改集合的操作，如果需要修改，应先使用ToArray()方法

public void ForeachDic()     
{     
     Dictionary<String, Int32> dic = new Dictionary<String, Int32>();   
     dic.Add("1", 10);     
     dic.Add("2", 20);     
     dic.Add("3", 30);     
     foreach (KeyValuePair<String, Int32> kvp in dic.ToArray())     
     {     
         Console.WriteLine(String.Format("Key:{0}; Value:{1}", kvp.Key, kvp.Value));     
         dic[kvp.Key] = 100;
     }     
}

2、把foreach改为 for(int i =0) 后，问题解决

private void ForeachDic()     
{     
     Dictionary<String, Int32> dic = new Dictionary<String, Int32>();   
     dic.Add("1", 10);     
     dic.Add("2", 20);     
     dic.Add("3", 30);    
     String[] keyArr = dic.Keys.ToArray<String>();     
     for (int i = 0; i < keyArr.Length; i++)     
     {     
         dic[keyStr[i]] = dic[keyStr[i]] + 1;     
      }     
} 

foreach是取只读的，在取的时候数据不能变（包括修改，删除，添加等）。要避免这个问题，就应该使用for循环

01，C# string类型转成byte[]：

Byte[] byteArray = System.Text.Encoding.Default.GetBytes ( str );

02， C# byte[]转成string：

stringstr = System.Text.Encoding.Default.GetString ( byteArray );

03，C# string类型转成ASCIIbyte[]：

（"01"转成byte[] =newbyte[]{0x30,0x31}）

byte[] byteArray = System.Text.Encoding.ASCII.GetBytes ( str );

04，C# ASCIIbyte[]转成string：

（byte[] = new byte[]{ 0x30, 0x31} 转成"01"）

string str = System.Text.Encoding.ASCII.GetString ( byteArray );

05， C# byte[]转16进制格式string：

new byte[]{ 0x30, 0x31}转成"3031":

public static string ToHexString ( byte[] bytes ) // 0xae00cf => "AE00CF "

{

string hexString = string.Empty;

if ( bytes != null )

{               

StringBuilder strB = new StringBuilder ();

for ( int i = 0; i < bytes.Length; i++ )

{                   

strB.Append ( bytes[i].ToString ( "X2" ) );               

}               

hexString = strB.ToString ();           

}return hexString;       

}

C# 常见的字节数组 byte[]截取

C# byte[]数组截取：

byte[] data= new byte[]{1,2,3,4,5,6,7,8,9};

截取2位：BitConverter.ToInt16(data,3)；//3表示从第三个位置开始

截取4位：BitConverter.ToInt32(data,3);

截取8位：BitConverter.ToInt64(data,3);

如果截取的数量不规则：data.Skip(5).Take(3).ToArray()；//表示从第五个位置截取三个字节

                                        data.Skip(4).ToArray();                //表示从第四个位置截取到最后

C# byte[]数组复制

(1)

byte[] data = new byte[]{0,1,2,3,4,5,6,7,8,9};

byte[] data1 = new byte[data.length];

for(int i=0;i<data1.length;i++){

        data1[i] = data[i];

}

(2)全部复制

byte[] data = new byte[]{0,1,2,3,4,5,6,7,8,9};

byte[] data1 = new byte[data.length];

Array.Copy(data,data1,data.length);//源数据data，目标数据data1，复制长度data.length

(3)全部复制

byte[] data = new byte[]{0,1,2,3,4,5,6,7,8,9};

byte[] data1;

data1 = (byte[])data .Clone();

(4)深度复制

byte[] srcArray = new byte[] { 0x01, 0x02, 0x03, 0x04 };

byte[] dstArray = new byte[srcArray.Length];

Buffer.BlockCopy(srcArray, 0, dstArray, 0, srcArray.Length);//源数据srcArray ， 起始位置0，目标数组dstArray ，开始位置0，                                                                                                      多少长度 srcArray.Length

C# byte[]数组删除

byte[] data = new byte[]{0,1,2,3,4,5,6,7,8,9};

(1)删除指定位置数量的数据

Array.Clear(data,0,5); //data 从起始位置0， 删除5个位置

(2)删除指定的数据--返回byte[] 空间减少

///

/// 去掉byte[] 中特定的byte

///

/// 需要处理的byte[]

/// byte[] 中需要除去的特定 byte (此处: byte cut = 0x00 ;)

///

public static byte[] ByteCut(byte[] b, byte cut)

{

var list = new List();

list.AddRange(b);

for (var i = list.Count - 1; i >= 0; i--)

{

if (list[i] == cut)

list.RemoveAt(i);

}

var lastbyte = new byte[list.Count];

for (var i = 0; i < list.Count; i++)

{

lastbyte[i] = list[i];

}

return lastbyte;

}

(3)清空byte[]

byte[] data = new byte[50];

data = new byte[50];//新new 出来则原来空间数据清除

配置文件如下,也是在网上找了很久的资料

打开nginx配置文件，nginx.conf

在http中加入以下代码：

Event

与信号量一样，事件也是一个系统范围内的资源同步方法。又分为ManualResetEvent，AutoResetEvent，CountdownEvent以及ManualResetEventSlim，在构建对象实例时若传入了name参数，代表这是一个可以跨进程的系统级同步事件。

以 ManualResetEvent 为例，该类有 signaled 和 nonsignaled 两种状态，这两种状态通过实例化对象时的 布尔类型 参数决定，TRUE 就是signaled, False相反

文档中常见的翻译是发出信号的状态和未发出信号的状态，微软官网的机翻是终止状态和非终止状态，还有一些释放线程之类的描述，直观上难以理解。其实就是改变状态而已，还不如英文的好理解。

ManualResetEvent 的基类 EventWaitHandle 中提供了Set() 和Reset() 方法，用于改变状态，Set 将事件修改为 signaled，Reset重置为 nonsignaled

这里说一下Set和Reset，这两个方法是改变了事件的状态，并不是一个瞬时性的动作，也就意味着在调用Set后，调用Reset之前，事件都处于 signaled 状态（AutoResetEvent会自动调用Reset重置事件状态）

WaitHandle 类中提供了众多等待信号的方法，EventWaitHandle 继承自WaitHandle， ManualResetEvent中也可以调用WaitOne等方法。

回头再看一下信号量 Semaphore 的示例，也调用 WaitOne 等待信号，因为它也继承自 Waithandler。

放到互联网一段时间试运行，会发现有人使用软件经常扫描我们的socket端口，并且造成了程序的卡死 ，今天通过优化，基本上解决了问题，但是我不断断开连接，导致rabbitmq崩掉了，无法连接，这个倒是没想到，周一在做进一步优化了，找找原因看到底是哪里的问题。

我这个项目主要是别人通过网页接口ajax请求启停设备，接口会将请求数据发送到rabbitmq，然后桌面应该程序接收到rabbitmq消息之后，通过4G向设备发送指令，今天所有指令需要发送两次，调试程序，也没发现问题，后来登录rabbitmq管理后台，发现多了一个连接，是之前服务器上安装的程序产生的长连接。之前已经关闭了应用程序，但是它却依然在连着rabbitmq，通过进程结束原程序，在本地测试，发送指令后，rabbitmq也及时的发送接收了。

但是这个问题很奇怪，可能队列之间有冲突吧，否则多个连接也不影响什么。

服务器上安装的版本是2019的SQL Server，我自己开发的电脑上安装的是2012的，将服务器上的分离之后拷贝到我的电脑上并不能附加到数据库中，发现是数据库版本不兼容，高版本的不能附加到低版本上面。就只能生成相应的脚本，建库建表，然后在导入数据进去。

步骤：

选择数据库，右击，“任务” 中选择 “生成脚本” ，

“下一步” ，到选择对象，第一个是数据库整体，第二个可自由选择数据库中的表，勾选其一，

然后 “下一步”，在“高级”中，将服务器版脚本更改到2012版的，然后在“要编写脚本的数据的类型”，选择 “架构和数据”。然后退回来，存储路径随意，然后在我的电脑上执行脚本就能使用了。

表的框架是搞定了，但是数据还要重新导入

在我的电脑上，点击数据库，选择任务-》导入数据，选择源数据 输入服务器地址 用户名和帐号，选择好数据库，设置本地可以直接用windows验证，选择数据库 ，下一步选择所有表，后面都直接下一步即可。 注意的是 如果表中有种子标识的，需要先将标识去掉，等数据全部导入完成后，在设置回来。

相反 如果低版本到高版本的，可以直接附加上去。

2.1 基本概念

提到RabbitMQ，就不得不提AMQP协议。AMQP协议是具有现代特征的二进制协议。是一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。

先了解一下AMQP协议中间的几个重要概念：

Server：接收客户端的连接，实现AMQP实体服务。
Connection：连接，应用程序与Server的网络连接，TCP连接。
Channel：信道，消息读写等操作在信道中进行。客户端可以建立多个信道，每个信道代表一个会话任务。
Message：消息，应用程序和服务器之间传送的数据，消息可以非常简单，也可以很复杂。由Properties和Body组成。Properties为外包装，可以对消息进行修饰，比如消息的优先级、延迟等高级特性；Body就是消息体内容。
Virtual Host：虚拟主机，用于逻辑隔离。一个虚拟主机里面可以有若干个Exchange和Queue，同一个虚拟主机里面不能有相同名称的Exchange或Queue。
Exchange：交换器，接收消息，按照路由规则将消息路由到一个或者多个队列。如果路由不到，或者返回给生产者，或者直接丢弃。RabbitMQ常用的交换器常用类型有direct、topic、fanout、headers四种，后面详细介绍。
Binding：绑定，交换器和消息队列之间的虚拟连接，绑定中可以包含一个或者多个RoutingKey。
RoutingKey：路由键，生产者将消息发送给交换器的时候，会发送一个RoutingKey，用来指定路由规则，这样交换器就知道把消息发送到哪个队列。路由键通常为一个“.”分割的字符串，例如“com.rabbitmq”。
Queue：消息队列，用来保存消息，供消费者消费。

2.2 工作原理

AMQP 协议模型由三部分组成：生产者、消费者和服务端，执行流程如下：

生产者是连接到 Server，建立一个连接，开启一个信道。
生产者声明交换器和队列，设置相关属性，并通过路由键将交换器和队列进行绑定。
消费者也需要进行建立连接，开启信道等操作，便于接收消息。
生产者发送消息，发送到服务端中的虚拟主机。
虚拟主机中的交换器根据路由键选择路由规则，发送到不同的消息队列中。
订阅了消息队列的消费者就可以获取到消息，进行消费。

2.3 常用交换器

RabbitMQ常用的交换器类型有direct、topic、fanout、headers四种：

Direct Exchange：见文知意，直连交换机意思是此交换机需要绑定一个队列，要求该消息与一个特定的路由键完全匹配。简单点说就是一对一的，点对点的发送。

Fanout Exchange：这种类型的交换机需要将队列绑定到交换机上。一个发送到交换机的消息都会被转发到与该交换机绑定的所有队列上。很像子网广播，每台子网内的主机都获得了一份复制的消息。简单点说就是发布订阅。

Topic Exchange：直接翻译的话叫做主题交换机，如果从用法上面翻译可能叫通配符交换机会更加贴切。这种交换机是使用通配符去匹配，路由到对应的队列。通配符有两种："*" 、 "#"。需要注意的是通配符前面必须要加上"."符号。

*符号：有且只匹配一个词。比如 a.*可以匹配到"a.b"、"a.c"，但是匹配不了"a.b.c"。

#符号：匹配一个或多个词。比如"rabbit.#"既可以匹配到"rabbit.a.b"、"rabbit.a"，也可以匹配到"rabbit.a.b.c"。

Headers Exchange：这种交换机用的相对没这么多。它跟上面三种有点区别，它的路由不是用routingKey进行路由匹配，而是在匹配请求头中所带的键值进行路由。创建队列需要设置绑定的头部信息，有两种模式：全部匹配和部分匹配。如上图所示，交换机会根据生产者发送过来的头部信息携带的键值去匹配队列绑定的键值，路由到对应的队列。

2.4 消费原理

我们先看几个基本概念：

broker：每个节点运行的服务程序，功能为维护该节点的队列的增删以及转发队列操作请求。
master queue：每个队列都分为一个主队列和若干个镜像队列。
mirror queue：镜像队列，作为master queue的备份。在master queue所在节点挂掉之后，系统把mirror queue提升为master queue，负责处理客户端队列操作请求。注意，mirror queue只做镜像，设计目的不是为了承担客户端读写压力。

集群中有两个节点，每个节点上有一个broker，每个broker负责本机上队列的维护，并且borker之间可以互相通信。集群中有两个队列A和B，每个队列都分为master queue和mirror queue（备份）。那么队列上的生产消费怎么实现的呢？

对于消费队列，有两个consumer消费队列A，这两个consumer连在了集群的不同机器上。RabbitMQ集群中的任何一个节点都拥有集群上所有队列的元信息，所以连接到集群中的任何一个节点都可以，主要区别在于有的consumer连在master queue所在节点，有的连在非master queue节点上。

因为mirror queue要和master queue保持一致，故需要同步机制，正因为一致性的限制，导致所有的读写操作都必须都操作在master queue上（想想，为啥读也要从master queue中读？和数据库读写分离是不一样的），然后由master节点同步操作到mirror queue所在的节点。即使consumer连接到了非master queue节点，该consumer的操作也会被路由到master queue所在的节点上，这样才能进行消费。

对于生成队列，原理和消费一样，如果连接到非 master queue 节点，则路由过去。

所以，到这里小伙伴们就可以看到 RabbitMQ的不足：由于master queue单节点，导致性能瓶颈，吞吐量受限。虽然为了提高性能，内部使用了Erlang这个语言实现，但是终究摆脱不了架构设计上的致命缺陷。

2.5 高级特性
2.5.1 过期时间

Time To Live，也就是生存时间，是一条消息在队列中的最大存活时间，单位是毫秒，下面看看RabbitMQ过期时间特性：

RabbitMQ可以对消息和队列设置TTL。
RabbitMQ支持设置消息的过期时间，在消息发送的时候可以进行指定，每条消息的过期时间可以不同。
RabbitMQ支持设置队列的过期时间，从消息入队列开始计算，直到超过了队列的超时时间配置，那么消息会变成死信，自动清除。
如果两种方式一起使用，则过期时间以两者中较小的那个数值为准。
当然也可以不设置TTL，不设置表示消息不会过期；如果设置为0，则表示除非此时可以直接将消息投递到消费者，否则该消息将被立即丢弃。

2.5.2 消息确认

为了保证消息从队列可靠地到达消费者，RabbitMQ提供了消息确认机制。

消费者订阅队列的时候，可以指定autoAck参数，当autoAck为true的时候，RabbitMQ采用自动确认模式，RabbitMQ自动把发送出去的消息设置为确认，然后从内存或者硬盘中删除，而不管消费者是否真正消费到了这些消息。

当autoAck为false的时候，RabbitMQ会等待消费者回复的确认信号，收到确认信号之后才从内存或者磁盘中删除消息。

消息确认机制是RabbitMQ消息可靠性投递的基础，只要设置autoAck参数为false，消费者就有足够的时间处理消息，不用担心处理消息的过程中消费者进程挂掉后消息丢失的问题。

2.5.3 持久化

消息的可靠性是RabbitMQ的一大特色，那么RabbitMQ是如何保证消息可靠性的呢？答案就是消息持久化。持久化可以防止在异常情况下丢失数据。RabbitMQ的持久化分为三个部分：交换器持久化、队列持久化和消息的持久化。

交换器持久化可以通过在声明队列时将durable参数设置为true。如果交换器不设置持久化，那么在RabbitMQ服务重启之后，相关的交换器元数据会丢失，不过消息不会丢失，只是不能将消息发送到这个交换器了。

队列的持久化能保证其本身的元数据不会因异常情况而丢失，但是不能保证内部所存储的消息不会丢失。要确保消息不会丢失，需要将其设置为持久化。队列的持久化可以通过在声明队列时将durable参数设置为true。

设置了队列和消息的持久化，当RabbitMQ服务重启之后，消息依然存在。如果只设置队列持久化或者消息持久化，重启之后消息都会消失。

当然，也可以将所有的消息都设置为持久化，但是这样做会影响RabbitMQ的性能，因为磁盘的写入速度比内存的写入要慢得多。

对于可靠性不是那么高的消息可以不采用持久化处理以提高整体的吞吐量。鱼和熊掌不可兼得，关键在于选择和取舍。在实际中，需要根据实际情况在可靠性和吞吐量之间做一个权衡。

2.5.4 死信队列

当消息在一个队列中变成死信之后，他能被重新发送到另一个交换器中，这个交换器成为死信交换器，与该交换器绑定的队列称为死信队列。

消息变成死信有下面几种情况：

消息被拒绝。

消息过期

队列达到最大长度

DLX也是一个正常的交换器，和一般的交换器没有区别，他能在任何的队列上面被指定，实际上就是设置某个队列的属性。当这个队列中有死信的时候，RabbitMQ会自动将这个消息重新发送到设置的交换器上，进而被路由到另一个队列，我们可以监听这个队列中消息做相应的处理。

死信队列有什么用？当发生异常的时候，消息不能够被消费者正常消费，被加入到了死信队列中。后续的程序可以根据死信队列中的内容分析当时发生的异常，进而改善和优化系统。

2.5.5 延迟队列

一般的队列，消息一旦进入队列就会被消费者立即消费。延迟队列就是进入该队列的消息会被消费者延迟消费，延迟队列中存储的对象是的延迟消息，“延迟消息”是指当消息被发送以后，等待特定的时间后，消费者才能拿到这个消息进行消费。

延迟队列用于需要延迟工作的场景。最常见的使用场景：淘宝或者天猫我们都使用过，用户在下单之后通常有30分钟的时间进行支付，如果这30分钟之内没有支付成功，那么订单就会自动取消。

除了延迟消费，延迟队列的典型应用场景还有延迟重试。比如消费者从队列里面消费消息失败了，可以延迟一段时间以后进行重试。

2.6 特性分析

这里才是内容的重点，不仅需要知道Rabbit的特性，还需要知道支持这些特性的原因：

消息路由（支持）：RabbitMQ可以通过不同的交换器支持不同种类的消息路由；
消息有序（不支持）：当消费消息时，如果消费失败，消息会被放回队列，然后重新消费，这样会导致消息无序；
消息时序（非常好）：通过延时队列，可以指定消息的延时时间，过期时间TTL等；
容错处理（非常好）：通过交付重试和死信交换器（DLX）来处理消息处理故障；
伸缩（一般）：伸缩其实没有非常智能，因为即使伸缩了，master queue还是只有一个，负载还是只有这一个master queue去抗，所以我理解RabbitMQ的伸缩很弱（个人理解）。
持久化（不太好）：没有消费的消息，可以支持持久化，这个是为了保证机器宕机时消息可以恢复，但是消费过的消息，就会被马上删除，因为RabbitMQ设计时，就不是为了去存储历史数据的。
消息回溯（不支持）：因为消息不支持永久保存，所以自然就不支持回溯。
高吞吐（中等）：因为所有的请求的执行，最后都是在master queue，它的这个设计，导致单机性能达不到十万级的标准。