【轉】Linux下消息隊列和socket絕對速度比拼

在當今的網絡時代，我們常常見到的進程間通信方式都是socket，比如Java的EJB調用，Java和C通信，Web Service服務等。socket是最常用的通訊技術，幾乎所有的系統、語言都支持，socket也是面向網絡的，通信的兩方可以跨越IP網絡進行傳輸。

在本地通信中(同一台機器上的進程間通訊)，socket的網絡特性卻成了累贅，組裝解析網絡報頭、報文確認、CRC校驗等都是針對網絡的，本地通信沒有必要，反而會影響傳輸效率。本地通信的一些傳統技術，如管道、FIFO、消息隊列等，沒有網絡功能的負擔，傳輸速度應該高於socket，那到底高多少以至於值得在應用中替換socket技術呢，今天就來一場小測試，就System V消息隊列和socket之間，做一次全面的速度比拼。

 

比拼場地

本人的筆記本：賽揚1.5G 內存1.5G
系統：Ubuntu8.04 Desktop (Linux 2.6.24-24-generic)
JDK：1.6

第一回合： Java測試

先說明一下，Java並不支持System V消息隊列，因此特為Java提供了JNI接口，我們使用lajp_9.09提供C源碼編譯的so動態連接庫，lajp的下載地址和文檔：http://code.google.com/p/lajp/

首先上場的是System V消息隊列。

發送端程序：

1. package test;
2.  
3. import lajp.MsgQ;
4.  
5. public class TestSend
6. {
7. /** 消息隊列KEY */
8. static final int IPC_KEY = 0×20021230;
9.  
10. static
11. {
12. //JNI
13.   System. loadLibrary("lajpmsgq");
14. }
15.  
16. public static void main(String[] args)
17. {
18. //創建或獲得現有的消息隊列
19. int msqid = MsgQ.msgget(IPC_KEY);
20. //發送字節數組
21. byte[] msg = new byte[1024];
22.  
23. for (int i = 0; i < 1024 * 5000; i++)
24. {
25. //每次發送1204字節到消息隊列，9527是消息類型
26. MsgQ. msgsnd(msqid, 9527, msg, msg.length);
27. }
28.  
29.   System. out.println("發送結束.");
30. }
31. }

 

接收端程序：

1. package test;
2.  
3. import lajp.MsgQ;
4.  
5. public class TestRcv
6. {
7. /** 消息隊列KEY */
8. static final int IPC_KEY = 0×20021230;
9.  
10. static
11. {
12. //JNI
13.   System. loadLibrary("lajpmsgq");
14. }
15.  
16. public static void main(String[] args)
17. {
18. //創建或獲得現有的消息隊列
19. int msqid = MsgQ.msgget(IPC_KEY);
20. //接收緩沖區
21. byte[] msg = new byte[1024];
22.  
23. long start = System.currentTimeMillis(); //開始時間
24.  
25. for (int i = 0; i < 1024 * 5000; i++)
26. {
27. //每次從消息隊列中接收消息類型為9527的消息，接收1204字節
28. MsgQ. msgrcv(msqid, msg, msg.length, 9527);
29. }
30.  
31. long end = System.currentTimeMillis(); //結束時間
32.   System. out.println("用時:" + (end – start) + "毫秒");
33. }
34. }

 

程序很簡單，需要說明的是三個JNI方法調用：

msgget()方法: System V消息隊列的技術要求，含義是通過一個指定的KEY獲得消息隊列標識符。
msgsnd()方法: 發送。
msgrcv()方法: 接收。

發送方進行了(1024 * 5000)次發送，每次發送1024字節數據，接收方進行了(1024 * 5000)次接收，每次接收1024字節，共計發送接收5G數據。測試時先啟動TestSend程序，再啟動TestRcv程序，共進行5輪次測試，測試結果如下：

用時:29846毫秒
用時:29591毫秒
用時:29935毫秒
用時:29730毫秒
用時:29468毫秒
平均速度：29714毫秒

用top命令監控測試期間的CPU、內存的使用：

接下來上場的是socket。

發送端程序：

1. import java.io.IOException;
2. import java.io.OutputStream;
3. import java.net.Socket;
4.  
5. public class SocketSend
6. {
7. public static void main(String[] args) throws IOException
8. {
9. //Socket
10.   Socket socket =  new Socket("127.0.0.1", 9527);
11. //輸出流
12.   OutputStream out = socket. getOutputStream();
13. //發送字節數組
14. byte[] msg = new byte[1024];
15.  
16. long start = System.currentTimeMillis(); //開始時間
17.  
18. for (int i = 0; i < 1024 * 5000; i++)
19. {
20. //發送
21. out. write(msg);
22. }
23.  
24. long end = System.currentTimeMillis(); //結束時間
25.   System. out.println("用時:" + (end – start) + "毫秒");
26. }
27. }

 

接收端程序：

1. import java.io.IOException;
2. import java.io.InputStream;
3. import java.net.ServerSocket;
4. import java.net.Socket;
5.  
6. public class SocketRecv
7. {
8. public static void main(String[] args) throws IOException
9. {
10. //偵聽9527端口
11.   ServerSocket serverSocket =  new ServerSocket(9527);
12. //Socket
13.   Socket socket = serverSocket. accept();
14. //輸入流
15.   InputStream in = socket. getInputStream();
16. //接收緩沖區
17. byte[] msg = new byte[1024];
18.  
19. for (int i = 0; i < 1024 * 5000; i++)
20. {
21. //每次接收1204字節
22. in. read(msg);
23. }
24.  
25.   System. out.println("接受結束.");
26. }
27. }

 

程序同樣很簡單，同樣發送接收了(1024 * 5000)次，同樣5G數據，socket程序必須先啟動服務方SocketRecv，然后啟動客戶方SocketSend，共進行5輪次測試，測試結果如下：

用時:33951毫秒
用時:33448毫秒
用時:33987毫秒
用時:34638毫秒
用時:33957毫秒
平均速度：33996.2毫秒

用top命令監控測試期間的CPU、內存的使用：

測試結果讓人對消息隊列有點失望，性能優勢微弱大約只領先了13%，且程序復雜性要大的多(使用了JNI)。不過重新審視測試過程有一個疑問：消息隊列程序調用了自定義的JNI接口，而socket是Java內嵌的功能，是否JVM對 socket有特殊的優化呢？

懷着這個疑問，進行第二場純C程序的測試。

第二回合： C程序測試

首先上場的還是System V消息隊列。

發送端程序：

1. #include <sys/ipc.h>
2. #include <sys/msg.h>
3. #include <stdio.h>
4.  
5. #define IPC_KEY 0×20021230 /* 消息隊列KEY */
6.  
7. /*消息結構*/
8. struct message
9. {
10. long msg_type; /* 消息標識符 */
11. char msg_text[1024]; /* 消息內容 */
12. };
13.  
14. int main()
15. {
16. /* 創建或獲得現有的消息隊列 */
17. int msqid = msgget(IPC_KEY, IPC_CREAT | 0666);
18. /* 消息結構 */
19. struct message msgq;
20. msgq. msg_type = 9527; /* 消息類型 */
21.  
22. int i;
23. for (i = 0; i < 1024 * 5000; i++)
24. {
25. /* 接收 */
26. msgsnd (msqid, &msgq, 1024, 0);
27. }
28.  
29.   printf ("msgq發送結束，共發送%d次\n", i);
30. return 0;
31. }

 

接收端程序：

1. #include <sys/ipc.h>
2. #include <sys/msg.h>
3. #include <stdio.h>
4.  
5. #define IPC_KEY 0×20021230 /* 消息隊列KEY */
6.  
7. /*消息結構*/
8. struct message
9. {
10. long msg_type; /* 消息標識符 */
11. char msg_text[1024]; /* 消息內容 */
12. };
13.  
14. int main()
15. {
16. /* 創建或獲得現有的消息隊列 */
17. int msqid = msgget(IPC_KEY, IPC_CREAT | 0666);
18. /* 消息結構 */
19. struct message msgq;
20.  
21. int i;
22. for (i = 0; i < 1024 * 5000; i++)
23. {
24. /* 接收 */
25. msgrcv (msqid, &msgq, 1024, 9527, 0);
26. }
27.  
28.   printf ("msgq接收結束，共接收%d次\n", i);
29. return 0;
30. }

 

和第一場一樣，發送接收了(1024 * 5000)次，同樣5G數據，先啟動接收端程序msgrecv，然后以$time msgsend方式啟動客戶端程序，共進行5輪次測試，time的測試結果如下：

用戶 系統 時鍾
第一次： 0.992s 7.084s 18.202s
第二次： 0.888s 7.280s 18.815s
第三次： 1.060s 7.656s 19.476s
第四次： 1.048s 7.124s 20.293s
第五次： 1.008s 7.160s 18.655s

用top命令監控測試期間的CPU、內存的使用：

接下來上場的是socket。

發送端程序：

1. #include <stdio.h>
2. #include <string.h>
3. #include <netdb.h>
4.  
5. char msg[1024]; /* 發送消息 */
6.  
7. int main()
8. {
9. char *ip = "127.0.0.1"; /* 發送地址 */
10. int port = 9527; /* 發送端口 */
11.  
12. struct hostent *server_host = gethostbyname(ip);
13.  
14. /* 客戶端填充 sockaddr 結構 */
15. struct sockaddr_in client_addr; /* 客戶端地址結構 */
16. bzero (&client_addr, sizeof(client_addr));
17. client_addr. sin_family = AF_INET; /* AF_INET:IPV4協議 */
18. client_addr. sin_addr.s_addr = ((struct in_addr *)(server_host->h_addr))->s_addr; /* 服務端地址 */
19. client_addr. sin_port = htons(port); /* 端口 */
20.  
21. /* 建立socket */
22. int sockfd = socket(AF_INET, SOCK_STREAM, 0);
23. /* 連接 */
24. connect (sockfd, (struct sockaddr *)(&client_addr), sizeof(client_addr));
25.  
26. int i;
27. for (i = 0; i < 1024 * 5000; i++)
28. {
29. /* 發送 */
30. send (sockfd, msg, 1024, 0);
31. }
32.  
33.   printf ("發送結束，共發送%d次\n", i);
34. return 0;
35. }

 

接收端程序：

1. #include <stdio.h>
2. #include <string.h>
3. #include <netdb.h>
4.  
5. char msg[1024]; /* 接收緩沖區 */
6.  
7. int main()
8. {
9. int listen_port = 9527; /* 偵聽端口 */
10. int listenfd = socket(AF_INET, SOCK_STREAM, 0); /* 建立偵聽socket */
11.  
12. /* 服務端填充 sockaddr 結構 */
13. struct sockaddr_in server_addr; /* 服務端地址結構 */
14. bzero (&server_addr, sizeof(server_addr));
15. server_addr. sin_family = AF_INET; /* AF_INET:IPV4協議 */
16. server_addr. sin_addr.s_addr = htonl(INADDR_ANY); /* INADDR_ANY:通配地址，表示內核選擇IP地址 */
17. server_addr. sin_port = htons(listen_port); /* 端口 */
18.  
19. /* 綁定端口 */
20. bind (listenfd, (struct sockaddr *)(&server_addr), sizeof(server_addr));
21. /* 偵聽 */
22. listen (listenfd, 5);
23. int sockfd = accept(listenfd, NULL, NULL);
24.  
25. int i;
26. for (i = 0; i < 1024 * 5000; i++)
27. {
28. /* 接收 */
29. recv (sockfd, msg, 1024, 0);
30. }
31.  
32.   printf ("接收結束，共接收%d次\n", i);
33. return 0;
34. }

 

C語言中，socket程序復雜了不少。測試標准和Java相同，發送接收了(1024 * 5000)次，5G數據，先啟動接收端程序，然后以time方式啟動發送端，測試結果如下：

用戶 系統 時鍾
第一次： 0.524s 9.765s 20.666s
第二次： 0.492s 9.825s 20.530s
第三次： 0.468s 9.493s 21.831s
第四次： 0.512s 9.205s 20.059s
第五次： 0.440s 9.605s 21.888s

用top命令監控測試期間的CPU、內存的使用：

C語言的socket程序系統用時多一些，消息隊列程序用戶用時多一些，這和他們的實現方式相關，從時鍾比較看，消息隊列比socket快10%左右，和Java測試結果相似。比較Java和C，C只領先了三分之一，看來當前的Java效率已經相當高了。

還不能忙於下結論，socket的通信方式一般有兩種：長連接和短連接。長連接指發送端和接收端建立連接后，可以保持socket通道進行多次消息傳輸，在這種場景基本不用計算socket建立和關閉的時間，前面的測試都是基於長連接方式；短連接一般在建立socket通道后，只進行一次通信，然后就關閉 socket通道，這種場景必須考慮socket建立和關閉的時間（socket建立連接需要三次握手，關閉連接要四次通信）。

第三回合： Java測試(短連接)

將第一回合中的Java程序稍作修改，先看socket的：

發送端程序：

1. import java.io.IOException;
2. import java.io.OutputStream;
3. import java.net.Socket;
4.  
5. public class SocketSend2
6. {
7. public static void main(String[] args) throws IOException
8. {
9. long start = System.currentTimeMillis(); //開始時間
10. //發送字節數組
11. byte[] msg = new byte[1024];
12.  
13. for (int i = 0; i < 1024 * 1000; i++)
14. {
15. //建立Socket連接
16.   Socket socket =  new Socket("127.0.0.1", 9527);
17. //輸出流
18.   OutputStream out = socket. getOutputStream();
19. //發送
20. out. write(msg);
21.  
22. //關閉輸出流
23. out. close();
24. //關閉socket連接
25. socket. close();
26. }
27.  
28. long end = System.currentTimeMillis(); //結束時間
29.   System. out.println("用時:" + (end – start) + "毫秒");
30. }
31. }

 

建立socket的語句放在了循環內部，這樣每次發送都是新建的連接，025行的關閉語句是必須的，因為socket是系統的有限資源，支持不了這么大規模的申請。

接收端程序：

1. import java.io.IOException;
2. import java.io.InputStream;
3. import java.net.ServerSocket;
4. import java.net.Socket;
5.  
6. public class SocketRecv2
7. {
8. public static void main(String[] args) throws IOException
9. {
10. //偵聽9527端口
11.   ServerSocket serverSocket =  new ServerSocket(9527);
12. //接收緩沖區
13. byte[] msg = new byte[1024];
14.  
15. for (int i = 0; i < 1024 * 1000; i++)
16. {
17. //接到客戶端Socket連接請求
18.   Socket socket = serverSocket. accept();
19. //輸入流
20.   InputStream in = socket. getInputStream();
21. //每次接收1204字節
22. in. read(msg);
23.  
24. //關閉輸入流
25. in. close();
26. //關閉socket連接
27. socket. close();
28. }
29.  
30.   System. out.println("接受結束.");
31. }
32. }

 

接收端也做了相應的改動，發送和接收次數降低到(1024 * 1000)次，測試結果：431280毫秒，不要吃驚，沒錯是431.280秒，這也是書本上為什么總在強調使用數據庫連接池的原因。

消息隊列沒有像socket那樣的連接概念，為了做個參考，將第一回合中的消息隊列程序也修改一下：

發送端程序：

1. package test;
2.  
3. import lajp.MsgQ;
4.  
5. public class TestSend2
6. {
7. /** 消息隊列KEY */
8. static final int IPC_KEY = 0×20021230;
9.  
10. static
11. {
12. //JNI
13.   System. loadLibrary("lajpmsgq");
14. }
15.  
16. public static void main(String[] args)
17. {
18. //發送字節數組
19. byte[] msg = new byte[1024];
20.  
21. for (int i = 0; i < 1024 * 1000; i++)
22. {
23. //創建或獲得現有的消息隊列
24. int msqid = MsgQ.msgget(IPC_KEY);
25.  
26. //每次發送1204字節
27. MsgQ. msgsnd(msqid, 9527, msg, msg.length);
28. }
29.  
30.   System. out.println("發送結束.");
31. }
32. }

 

將024行的msgget()方法放在循環內部，作為和socket比較的“連接”。

接收段程序：

1. package test;
2.  
3. import lajp.MsgQ;
4.  
5. public class TestRcv2
6. {
7. /** 消息隊列KEY */
8. static final int IPC_KEY = 0×20021230;
9.  
10. static
11. {
12. //JNI
13.   System. loadLibrary("lajpmsgq");
14. }
15.  
16. public static void main(String[] args)
17. {
18. long start = System.currentTimeMillis(); //開始時間
19. //接收緩沖區
20. byte[] msg = new byte[1024];
21.  
22. for (int i = 0; i < 1024 * 1000; i++)
23. {
24. //創建或獲得現有的消息隊列
25. int msqid = MsgQ.msgget(IPC_KEY);
26.  
27. //每次接收1204字節
28. MsgQ. msgrcv(msqid, msg, msg.length, 9527);
29. }
30.  
31. long end = System.currentTimeMillis(); //結束時間
32.   System. out.println("用時:" + (end – start) + "毫秒");
33. }
34. }

 

測試結果：6617毫秒。

總結：

在能夠使用socket長連接的應用中，建議使用socket技術，畢竟很通用熟悉的人也多，而消息隊列能夠提高的效率有限；在只能使用socket短連接的應用中，特別是並發量大的場景，強烈建議使用消息隊列，因為能夠極大的提高通信速率。

 

（長連接指發送端和接收端建立連接后，可以保持socket通道進行多次消息傳輸，在這種場景基本不用計算socket建立和關閉的時間，前面的測試都是基於長連接方式；短連接一般在建立socket通道后，只進行一次通信，然后就關閉 socket通道，這種場景必須考慮socket建立和關閉的時間（socket建立連接需要三次握手，關閉連接要四次通信）。）