linux消息隊列命令 linux 消息隊列 系統(tǒng)參數(shù)

Linux 消息隊列長度處理

問題：

創(chuàng)新互聯(lián)公司是專業(yè)的黑山網(wǎng)站建設(shè)公司，黑山接單;提供網(wǎng)站設(shè)計、成都網(wǎng)站制作,網(wǎng)頁設(shè)計,網(wǎng)站設(shè)計,建網(wǎng)站,PHP網(wǎng)站建設(shè)等專業(yè)做網(wǎng)站服務(wù);采用PHP框架,可快速的進(jìn)行黑山網(wǎng)站開發(fā)網(wǎng)頁制作和功能擴展;專業(yè)做搜索引擎喜愛的網(wǎng)站,專業(yè)的做網(wǎng)站團(tuán)隊,希望更多企業(yè)前來合作!

在Linux 系統(tǒng)中通過消息隊列進(jìn)行進(jìn)程間的通訊時，只要定義的BufSize小于1024，隊列就能正常讀寫，當(dāng)Size定義大于1024時，隊列就無法成功。

處理步驟：

SystemV的消息隊列

/etc/sysctl.conf

修改

kernel.msgmni=1000

kernel.msgmax=81920

kernel.msgmnb=163840

msgmni為MSGMNI，即系統(tǒng)的消息隊列數(shù)目。平臺每個DTA需要使用3個消息隊列，即最大DTA數(shù)為1000/3。該參數(shù)應(yīng)該比平臺最大隊列個數(shù)參數(shù)配置大。

msgmax為MSGMAX，即一個消息的字節(jié)大小。目前擴展值為8k，平臺一個交易消息為4個字節(jié)，不會超過限制。

msgmnb為MSGMNB，即隊列存放消息的總字節(jié)數(shù)。

POSIX消息隊列

修改

fs.mqueue.msg_max=1000??-消息個數(shù)

fs.?mqueue.?msgsize_max=8192?-消息長度

另外操作系統(tǒng)對文件大小的限制ulimit?-q你可以看到POSIX消息隊列的最大容量

cat?/proc/sys/kernel/msgmax

cat?/proc/sys/kernel/msgmni

cat?/proc/sys/kernel/msgmnb

linux消息隊列的收發(fā)

26203就是你這個程序運行時候的 PID 阿，你在發(fā)消息之前，不是把當(dāng)前進(jìn)程的PID存放到消息里的 mtype 了嗎

p-mtype = getpid();

然后接收方收到這個消息，又把 mtype 打印出來了阿

printf("Message received from %ld\n%s\n" , p-mtype , p-mdata );

這個就是打印從 PID 收到了消息，消息是 p-mdata里的內(nèi)容。

Linux系統(tǒng)編程—消息隊列

消息隊列本質(zhì)上是位于內(nèi)核空間的鏈表，鏈表的每個節(jié)點都是一條消息。每一條消息都有自己的消息類型，消息類型用整數(shù)來表示，而且必須大于 0。每種類型的消息都被對應(yīng)的鏈表所維護(hù)：

其中數(shù)字 1 表示類型為 1 的消息，數(shù)字2、3、4 類似。彩色塊表示消息數(shù)據(jù)，它們被掛在對應(yīng)類型的鏈表上。

值得注意的是，剛剛說過沒有消息類型為 0 的消息，實際上，消息類型為 0 的鏈表記錄了所有消息加入隊列的順序，其中紅色箭頭表示消息加入的順序。

無論你是發(fā)送還是接收消息，消息的格式都必須按照規(guī)范來。簡單的說，它一般長成下面這個樣子：

所以，只要你保證首4字節(jié)(32 位 linux 下的 long)是一個整數(shù)就行了。

舉個例子：

從上面可以看出，正文部分是什么數(shù)據(jù)類型都沒關(guān)系，因為消息隊列傳遞的是 2 進(jìn)制數(shù)據(jù)，不一定非得是文本。

msgsnd 函數(shù)用于將數(shù)據(jù)發(fā)送到消息隊列。如果該函數(shù)被信號打斷，會設(shè)置 errno 為 EINTR。

參數(shù) msqid：ipc 內(nèi)核對象 id

參數(shù) msgp：消息數(shù)據(jù)地址

參數(shù) msgsz：消息正文部分的大?。ú话㈩愋停?/p>

參數(shù) msgflg：可選項

該值為 0：如果消息隊列空間不夠，msgsnd 會阻塞。

IPC_NOWAIT：直接返回，如果空間不夠，會設(shè)置 errno 為 EAGIN.

返回值：0 表示成功，-1 失敗并設(shè)置 errno。

msgrcv 函數(shù)從消息隊列取出消息后，并將其從消息隊列里刪除。

參數(shù) msqid：ipc 內(nèi)核對象 id

參數(shù) msgp：用來接收消息數(shù)據(jù)地址

參數(shù) msgsz：消息正文部分的大?。ú话㈩愋停?/p>

參數(shù) msgtyp：指定獲取哪種類型的消息

msgtyp = 0：獲取消息隊列中的第一條消息

msgtyp 0：獲取類型為 msgtyp 的第一條消息，除非指定了 msgflg 為MSG_EXCEPT，這表示獲取除了 msgtyp 類型以外的第一條消息。

msgtyp 0：獲取類型 ≤|msgtyp|≤|msgtyp| 的第一條消息。

參數(shù) msgflg：可選項。

如果為 0 表示沒有消息就阻塞。

IPC_NOWAIT：如果指定類型的消息不存在就立即返回，同時設(shè)置 errno 為 ENOMSG

MSG_EXCEPT：僅用于 msgtyp 0 的情況。表示獲取類型不為 msgtyp 的消息

MSG_NOERROR：如果消息數(shù)據(jù)正文內(nèi)容大于 msgsz，就將消息數(shù)據(jù)截斷為 msgsz

程序 msg_send 和 msg_recv 分別用于向消息隊列發(fā)送數(shù)據(jù)和接收數(shù)據(jù)。

msg_send 程序定義了一個結(jié)構(gòu)體 Msg，消息正文部分是結(jié)構(gòu)體 Person。該程序向消息隊列發(fā)送了 10 條消息。

msg_send.c

程序 msg_send 第一次運行完后，內(nèi)核中的消息隊列大概像下面這樣：

msg_recv 程序接收一個參數(shù)，表示接收哪種類型的消息。比如./msg_recv 4 表示接收類型為 4 的消息，并打印在屏幕。

先運行 msg_send，再運行 msg_recv。

接收所有消息

接收類型為 4 的消息

獲取和設(shè)置消息隊列的屬性

msqid：消息隊列標(biāo)識符

cmd：控制指令

IPC_STAT:獲得msgid的消息隊列頭數(shù)據(jù)到buf中

IPC_SET：設(shè)置消息隊列的屬性，要設(shè)置的屬性需先存儲在buf中，可設(shè)置的屬性包括：msg_perm.uid、msg_perm.gid、msg_perm.mode以及msg_qbytes

buf：消息隊列管理結(jié)構(gòu)體。

返回值：

成功：0

出錯：-1，錯誤原因存于error中

EACCESS：參數(shù)cmd為IPC_STAT，確無權(quán)限讀取該消息隊列

EFAULT：參數(shù)buf指向無效的內(nèi)存地址

EIDRM：標(biāo)識符為msqid的消息隊列已被刪除

EINVAL：無效的參數(shù)cmd或msqid

EPERM：參數(shù)cmd為IPC_SET或IPC_RMID，卻無足夠的權(quán)限執(zhí)行

自己實現(xiàn)消息隊列msg queue linux C

因為不僅僅信號量，共享內(nèi)存、消息隊列在NDK下都不能用，所以之前使用Linux 下IPC的消息隊列，msgget/msgsnd/msgrcv都不能使用，所以沒有辦法，只能自己實現(xiàn)消息隊列，采用linux 下互斥鎖和條件變量實現(xiàn)了讀時-隊列空-會阻塞，寫時-隊列滿-會阻塞。

talk is easy, show me the code. -- 廢話少說，放碼過來。編譯時候使用?cc main.c -pthread，注意-pthread參數(shù)，因為依賴線程庫。

########################################################################

#include stdio.h

#include pthread.h

#include string.h

#include stdlib.h

#includesys/time.h

#define MAX_QUEUE_SIZE_IN_BYTES (1024)

#define MQ_SIZE_MAX 512

#define MQ_LENGTH_MAX 30

#define MQ_NAME "msg queue example"

typedef struct _simple_queue

{

int front;

int rear;

int length;

int queue_type;

pthread_mutex_t data_mutex;

pthread_cond_t data_cond;

int write_pos;

char queue_name[32];

void *data[0];

}simple_queue;

typedef enum _queue_type

{

QUEUE_BLOCK = 0,

QUEUE_NO_BLOCK,

}queue_type;

typedef enum _queue_status

{

QUEUE_IS_NORMAL = 0,

QUEUE_NO_EXIST,

QUEUE_IS_FULL,

QUEUE_IS_EMPTY,

}queue_status;

typedef enum _cntl_queue_ret

{

CNTL_QUEUE_SUCCESS = 0,

CNTL_QUEUE_FAIL,

CNTL_QUEUE_TIMEOUT,

CNTL_QUEUE_PARAM_ERROR,

}cntl_queue_ret;

typedef enum _queue_flag

{

IPC_BLOCK = 0,

IPC_NOWAIT = 1,

IPC_NOERROR = 2,

}queue_flag;

typedef struct _simple_queue_buf

{

int msg_type;

char msg_buf[50];

}queue_buf;

simple_queue* create_simple_queue(const char* queue_name, int queue_length, int queue_type)

{

simple_queue *this = NULL;

if (NULL == queue_name || 0 == queue_length)

{

printf("[%s] param is error\n", __FUNCTION__);

return NULL;

}

if(queue_length MAX_QUEUE_SIZE_IN_BYTES)

{

printf("[%s] param is error,queue_length should less than %d bytes\n", __FUNCTION__, MAX_QUEUE_SIZE_IN_BYTES);

return NULL;

}

this = (simple_queue*)malloc(sizeof(simple_queue) + queue_length * sizeof(void*));

if (NULL != this)

{

this-front = 0;

this-rear = 0;

this-length = queue_length;

this-queue_type = queue_type;

if (0 != pthread_mutex_init((this-data_mutex), NULL) || 0 != pthread_cond_init((this-data_cond), NULL))

{

printf("[%s]pthread_mutex_init failed!\n", __FUNCTION__);

free(this);

this = NULL;

return NULL;

}

strcpy(this-queue_name, queue_name);

}

else

{

printf("[%s]malloc is failed!\n", __FUNCTION__);

return NULL;

}

return this;

}

queue_status is_full_queue(simple_queue* p_queue)

{

queue_status ret = QUEUE_IS_NORMAL;

do

{

if (NULL == p_queue)

{

printf("[%s] param is error\n", __FUNCTION__);

ret = QUEUE_NO_EXIST;

break;

}

if (p_queue-front == ((p_queue-rear + 1) % (p_queue-length)))

{

printf("[%s] queue is full\n", __FUNCTION__);

ret = QUEUE_IS_FULL;

break;

}

}while(0);

return ret;

}

queue_status is_empty_queue(simple_queue* p_queue)

{

queue_status ret = QUEUE_IS_NORMAL;

do

{

if (NULL == p_queue)

{

printf("[%s] param is error\n", __FUNCTION__);

ret = QUEUE_NO_EXIST;;

break;

}

if (p_queue-front == p_queue-rear)

{

printf("[%s] queue is empty\n", __FUNCTION__);

ret = QUEUE_IS_EMPTY;

break;

}

}while(0);

return ret;

}

cntl_queue_ret push_simple_queue(simple_queue* p_queue, void* data, queue_flag flg)

{

int w_cursor = 0;

if(NULL == p_queue || NULL == data)

{

printf("[%s] param is error\n", __FUNCTION__);

return CNTL_QUEUE_PARAM_ERROR;

}

pthread_mutex_lock((p_queue-data_mutex));

w_cursor = (p_queue-rear + 1)%p_queue-length;

if (w_cursor == p_queue-front)

{

if(flg == IPC_BLOCK)

{

pthread_cond_wait((p_queue-data_cond), (p_queue-data_mutex));

}

else

{

printf("[%s]: queue is full\n", __FUNCTION__);

pthread_mutex_unlock((p_queue-data_mutex));

return CNTL_QUEUE_FAIL;

}

w_cursor = (p_queue-rear + 1)%p_queue-length;

}

p_queue-data[p_queue-rear] = data;

p_queue-rear = w_cursor;

pthread_mutex_unlock((p_queue-data_mutex));

pthread_cond_signal((p_queue-data_cond));

return CNTL_QUEUE_SUCCESS;

}

cntl_queue_ret pop_simple_queue(simple_queue* p_queue, void** data, queue_flag flg)

{

if(NULL == p_queue)

{

printf("[%s] param is error\n", __FUNCTION__);

return CNTL_QUEUE_PARAM_ERROR;

}

pthread_mutex_lock((p_queue-data_mutex));

if (p_queue-front == p_queue-rear)

{

if(flg == IPC_BLOCK)

{

pthread_cond_wait((p_queue-data_cond), (p_queue-data_mutex));

}

else

{

printf("[%s]: queue is empty\n", __FUNCTION__);

pthread_mutex_unlock((p_queue-data_mutex));

return CNTL_QUEUE_FAIL;

}

}

*data = p_queue-data[p_queue-front];

p_queue-front = (p_queue-front + 1)%p_queue-length;

pthread_mutex_unlock((p_queue-data_mutex));

pthread_cond_signal((p_queue-data_cond));

return CNTL_QUEUE_SUCCESS;

}

cntl_queue_ret destroy_simple_queue(simple_queue* p_queue)

{

cntl_queue_ret ret = CNTL_QUEUE_SUCCESS;

if(NULL == p_queue)

{

printf("[%s] param is error\n", __FUNCTION__);

ret = CNTL_QUEUE_PARAM_ERROR;

}

else

{

pthread_mutex_destroy((p_queue-data_mutex));

pthread_cond_destroy((p_queue-data_cond));

while (p_queue-front != p_queue-rear)//刪除隊列中殘留的消息

{

free(p_queue-data[p_queue-front]);

p_queue-front = (p_queue-front + 1)%p_queue-length;

}

free(p_queue);

p_queue = NULL;

}

return ret;

}

void* send_msg_thread(void* arg)

{

queue_buf* send_buf = NULL;

int i;

send_buf = (queue_buf*)malloc(sizeof(queue_buf));

send_buf-msg_type = 1;

strcpy(send_buf-msg_buf, "hello, world!");

printf("first1: rear =%d font =%d\n", ((simple_queue*)arg)-rear, ((simple_queue*)arg)-front);

if (push_simple_queue((simple_queue*)arg, (void*)send_buf, IPC_BLOCK) 0)

{

? ? printf("[%s]: push_simple_queue\n", __FUNCTION__);?

? ? return NULL;

}

printf("first2: rear =%d font =%d\n", ((simple_queue*)arg)-rear, ((simple_queue*)arg)-front);

queue_buf* send_buf1 = NULL;

send_buf1 = (queue_buf*)malloc(sizeof(queue_buf));

send_buf1-msg_type = 2;

strcpy(send_buf1-msg_buf, "byebye");

printf("first1: rear =%d font =%d\n", ((simple_queue*)arg)-rear, ((simple_queue*)arg)-front);

if (push_simple_queue((simple_queue*)arg, (void*)send_buf1, IPC_NOWAIT) 0)

{

? ? printf("[%s]: push_simple_queue\n", __FUNCTION__);?

? ? return NULL;

}

printf("first2: rear =%d font =%d\n", ((simple_queue*)arg)-rear, ((simple_queue*)arg)-front);

return NULL;

}

void* recv_msg_thread(void* arg)

{

int i;

queue_buf* recv_buf = (queue_buf*)malloc(sizeof(queue_buf));

printf("second1 rear =%d font =%d\n", ((simple_queue*)arg)-rear, ((simple_queue*)arg)-front);

if (CNTL_QUEUE_SUCCESS != pop_simple_queue((simple_queue*)arg, (void**)recv_buf, IPC_BLOCK))

{?

? ? printf("[%s]: pop_simple_queue failed!\n", __FUNCTION__);

return NULL;

}

for(i=0; i50; i++)

printf("%c", recv_buf-msg_buf[i]);

printf("\r\n");

printf("second2: rear =%d font =%d\n", ((simple_queue*)arg)-rear, ((simple_queue*)arg)-front);

printf("second1: rear =%d font =%d\n", ((simple_queue*)arg)-rear, ((simple_queue*)arg)-front);

if (CNTL_QUEUE_SUCCESS != pop_simple_queue((simple_queue*)arg, (void**)recv_buf, IPC_NOWAIT))

{?

? ? printf("[%s]: pop_simple_queue failed!\n", __FUNCTION__);

return NULL;

}

for(i=0; i50; i++)

printf("%c", recv_buf-msg_buf[i]);

printf("\r\n");

printf("second2: rear =%d font =%d\n", ((simple_queue*)arg)-rear, ((simple_queue*)arg)-front);

free(recv_buf);

recv_buf = NULL;

return NULL;

}

int main(int argc, char* argv[])

{

int ret = 0;

pthread_t send_thread_id = 0;

pthread_t recv_thread_id = 0;

simple_queue* msg_queue = NULL;

msg_queue = create_simple_queue(MQ_NAME, MQ_LENGTH_MAX, QUEUE_NO_BLOCK);

if (NULL == msg_queue)

{

printf("[%s]: create simple queue failed!\n", __FUNCTION__);

return -1;

}

ret = pthread_create(send_thread_id, NULL, send_msg_thread, (void*)msg_queue);

if (0 != ret)

{

printf("[%s]: create send thread failed!\n", __FUNCTION__);

return -1;

}

ret = pthread_create(recv_thread_id, NULL, recv_msg_thread, (void*)msg_queue);

if (0 != ret)

{

printf("[%s]: create recv thread failed!\n", __FUNCTION__);

return -1;

}

printf("begin join\n");

pthread_join(send_thread_id, NULL);

pthread_join(recv_thread_id, NULL);

printf("end join\n");

ret = destroy_simple_queue(msg_queue);

if (CNTL_QUEUE_SUCCESS != ret)

{

printf("[%s]: destroy simple queue failed!\n", __FUNCTION__);

return -1;

}

return 0;

}

文章題目：linux消息隊列命令 linux 消息隊列 系統(tǒng)參數(shù)
當(dāng)前URL：http://muchs.cn/article0/doshpoo.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供外貿(mào)網(wǎng)站建設(shè)、建站公司、品牌網(wǎng)站建設(shè)、網(wǎng)站改版、面包屑導(dǎo)航、

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時需注明來源： 創(chuàng)新互聯(lián)