消息隊列本質上是位于內核空間的鏈表，鏈表的每個節(jié)點都是一條消息。每一條消息都有自己的消息類型，消息類型用整數來表示，而且必須大于 0。每種類型的消息都被對應的鏈表所維護：
創(chuàng)新互聯主營鐵山港網站建設的網絡公司,主營網站建設方案,重慶App定制開發(fā),鐵山港h5微信小程序搭建,鐵山港網站營銷推廣歡迎鐵山港等地區(qū)企業(yè)咨詢
其中數字 1 表示類型為 1 的消息，數字2、3、4 類似。彩色塊表示消息數據，它們被掛在對應類型的鏈表上。
值得注意的是，剛剛說過沒有消息類型為 0 的消息，實際上，消息類型為 0 的鏈表記錄了所有消息加入隊列的順序，其中紅色箭頭表示消息加入的順序。
無論你是發(fā)送還是接收消息，消息的格式都必須按照規(guī)范來。簡單的說，它一般長成下面這個樣子：
所以，只要你保證首4字節(jié)(32 位 linux 下的 long)是一個整數就行了。
舉個例子：
從上面可以看出，正文部分是什么數據類型都沒關系，因為消息隊列傳遞的是 2 進制數據，不一定非得是文本。
msgsnd 函數用于將數據發(fā)送到消息隊列。如果該函數被信號打斷，會設置 errno 為 EINTR。
參數 msqid：ipc 內核對象 id
參數 msgp：消息數據地址
參數 msgsz：消息正文部分的大小（不包含消息類型）
參數 msgflg：可選項
該值為 0：如果消息隊列空間不夠，msgsnd 會阻塞。
IPC_NOWAIT：直接返回，如果空間不夠，會設置 errno 為 EAGIN.
返回值：0 表示成功，-1 失敗并設置 errno。
msgrcv 函數從消息隊列取出消息后，并將其從消息隊列里刪除。
參數 msqid：ipc 內核對象 id
參數 msgp：用來接收消息數據地址
參數 msgsz：消息正文部分的大?。ú话㈩愋停?/p>
參數 msgtyp：指定獲取哪種類型的消息
msgtyp = 0：獲取消息隊列中的第一條消息
msgtyp 0：獲取類型為 msgtyp 的第一條消息，除非指定了 msgflg 為MSG_EXCEPT，這表示獲取除了 msgtyp 類型以外的第一條消息。
msgtyp 0：獲取類型 ≤|msgtyp|≤|msgtyp| 的第一條消息。
參數 msgflg：可選項。
如果為 0 表示沒有消息就阻塞。
IPC_NOWAIT：如果指定類型的消息不存在就立即返回，同時設置 errno 為 ENOMSG
MSG_EXCEPT：僅用于 msgtyp 0 的情況。表示獲取類型不為 msgtyp 的消息
MSG_NOERROR：如果消息數據正文內容大于 msgsz，就將消息數據截斷為 msgsz
程序 msg_send 和 msg_recv 分別用于向消息隊列發(fā)送數據和接收數據。
msg_send 程序定義了一個結構體 Msg，消息正文部分是結構體 Person。該程序向消息隊列發(fā)送了 10 條消息。
msg_send.c
程序 msg_send 第一次運行完后，內核中的消息隊列大概像下面這樣：
msg_recv 程序接收一個參數，表示接收哪種類型的消息。比如./msg_recv 4 表示接收類型為 4 的消息，并打印在屏幕。
先運行 msg_send，再運行 msg_recv。
接收所有消息
接收類型為 4 的消息
獲取和設置消息隊列的屬性
msqid：消息隊列標識符
cmd：控制指令
IPC_STAT:獲得msgid的消息隊列頭數據到buf中
IPC_SET：設置消息隊列的屬性，要設置的屬性需先存儲在buf中，可設置的屬性包括：msg_perm.uid、msg_perm.gid、msg_perm.mode以及msg_qbytes
buf：消息隊列管理結構體。
返回值：
成功：0
出錯：-1，錯誤原因存于error中
EACCESS：參數cmd為IPC_STAT，確無權限讀取該消息隊列
EFAULT：參數buf指向無效的內存地址
EIDRM：標識符為msqid的消息隊列已被刪除
EINVAL：無效的參數cmd或msqid
EPERM：參數cmd為IPC_SET或IPC_RMID，卻無足夠的權限執(zhí)行
因為不僅僅信號量，共享內存、消息隊列在NDK下都不能用，所以之前使用Linux 下IPC的消息隊列，msgget/msgsnd/msgrcv都不能使用，所以沒有辦法，只能自己實現消息隊列，采用linux 下互斥鎖和條件變量實現了讀時-隊列空-會阻塞，寫時-隊列滿-會阻塞。
talk is easy, show me the code. -- 廢話少說，放碼過來。編譯時候使用?cc main.c -pthread，注意-pthread參數，因為依賴線程庫。
########################################################################
#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 系統(tǒng)中通過消息隊列進行進程間的通訊時，只要定義的BufSize小于1024，隊列就能正常讀寫，當Size定義大于1024時，隊列就無法成功。
處理步驟：
SystemV的消息隊列
/etc/sysctl.conf
修改
kernel.msgmni=1000
kernel.msgmax=81920
kernel.msgmnb=163840
msgmni為MSGMNI，即系統(tǒng)的消息隊列數目。平臺每個DTA需要使用3個消息隊列，即最大DTA數為1000/3。該參數應該比平臺最大隊列個數參數配置大。
msgmax為MSGMAX，即一個消息的字節(jié)大小。目前擴展值為8k，平臺一個交易消息為4個字節(jié)，不會超過限制。
msgmnb為MSGMNB，即隊列存放消息的總字節(jié)數。
POSIX消息隊列
修改
fs.mqueue.msg_max=1000??-消息個數
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
#include
#include
#include
#include types.h
#include msg.h
#include
#include ipc.h
void msg_show_attr(int msg_id, struct msqid_ds msg_info)
{
int ret = -1;
sleep(1);
ret = msgctl(msg_id, IPC_STAT, msg_info);
if( -1 == ret)
{
printf(獲消息信息失敗\n);
return ;
}
printf(\n);
printf(現隊列字節(jié)數：%d\n,msg_info.msg_cbytes);
printf(隊列消息數：%d\n,msg_info.msg_qnum);
printf(隊列字節(jié)數：%d\n,msg_info.msg_qbytes);
printf(發(fā)送消息進程pid：%d\n,msg_info.msg_lspid);
printf(接收消息進程pid：%d\n,msg_info.msg_lrpid);
printf(發(fā)送消息間：%s,ctime((msg_info.msg_stime)));
printf(接收消息間：%s,ctime((msg_info.msg_rtime)));
printf(變化間：%s,ctime((msg_info.msg_ctime)));
printf(消息UID：%d\n,msg_info.msg_perm.uid);
printf(消息GID：%d\n,msg_info.msg_perm.gid);
}
int main(void)
{
int ret = -1;
int msg_flags, msg_id;
key_t key;
struct msgmbuf{
int mtype;
char mtext[10];
};
struct msqid_ds msg_info;
struct msgmbuf msg_mbuf;
int msg_sflags,msg_rflags;
char *msgpath = /ipc/msg/;
key = ftok(msgpath,’a');
if(key != -1)
{
printf(功建立KEY\n);
}
else
{
printf(建立KEY失敗\n);
}
msg_flags = IPC_CREAT;
msg_id = msgget(key, msg_flags|0666);
if( -1 == msg_id)
{
printf(消息建立失敗\n);
return 0;
}
msg_show_attr(msg_id, msg_info);
msg_sflags = IPC_NOWAIT;
msg_mbuf.mtype = 10;
memcpy(msg_mbuf.mtext,測試消息,sizeof(測試消息));
ret = msgsnd(msg_id, msg_mbuf, sizeof(測試消息), msg_sflags);
if( -1 == ret)
{
printf(發(fā)送消息失敗\n);
}
msg_show_attr(msg_id, msg_info);
msg_rflags = IPC_NOWAIT|MSG_NOERROR;
ret = msgrcv(msg_id, msg_mbuf, 10,10,msg_rfla
共享內存示例代碼：
#include
#include sem.h
#include ipc.h
#include
typedef int sem_t;
union semun {
int val;
struct semid_ds *buf;
unsigned short *array;
} arg;
sem_t CreateSem(key_t key, int value)
{
union semun sem;
sem_t semid;
sem.val = value;
semid = semget(key,value,IPC_CREAT|0666);
if (-1 == semid)
{
printf(create semaphore error\n);
return -1;
}
semctl(semid,0,SETVAL,sem);
return semid;
}
/*
struct sembuf{
ushort sem_num;
short sem_op;
short sem_flg;
};
*/
void SetvalueSem(sem_t semid, int value)
{
union semun sem;
sem.val = value;
semctl(semid,0,SETVAL,sem);
return ;
}
int GetvalueSem(sem_t semid)
{
union semun sem;
return semctl(semid,0,GETVAL,sem);
return sem.val;
}
void DestroySem(sem_t semid)
{
union semun sem;
sem.val = 0;
semctl(semid,0,IPC_RMID,sem);
}
int Sem_P(sem_t semid)
{
struct sembuf sops={0,+1,IPC_NOWAIT};
return (semop(semid,sops,1));
}
int Sem_V(sem_t semid)
{
struct sembuf sops={0,-1,IPC_NOWAIT};
return (semop(semid,sops,1));
}
static char msg[]=共享內存\n;
int main(void)
{
key_t key;
int semid,shmid;
char i,*shms,*shmc;
struct semid_ds buf;
int value = 0;
char buffer[80];
pid_t p;
key = ftok(/ipc/sem/,’a');
shmid = shmget(key,1024,IPC_CREAT|0604);
semid = CreateSem(key,1);
p = fork();
if(p 0)
{
/* 父進程 */
/* 建立共享內存 */
shms = (char *)shmat(shmid,0,0);
memcpy(shms, msg, strlen(msg)+1);
sleep(10);
Sem_P(semid);
shmdt(shms);
DestroySem(semid);
}
else if(p == 0)
{
shmc = (char *)shmat(shmid,0,0);
Sem_V(semid);
printf(共享內存值:%s\n,shmc);
shmdt(sg_
重慶分公司
重慶分公司
