其中数字 1 表示类型为 1 的消息，数字2、3、4 类似。彩色块表示消息数据，它们被挂在对应类型的链表上。
值得注意的是，刚刚说过没有消息类型为 0 的消息，实际上，消息类型为 0 的链表记录了所有消息加入队列的顺序，其中红色箭头表示消息加入的顺序。
无论你是发送还是接收消息，消息的格式都必须按照规范来。简单的说，它一般长成下面这个样子：
所以，只要你保证首4字节(32 位 linux 下的 long)是一个整数就行了。
举个例子：
从上面可以看出，正文部分是什么数据类型都没关系，因为消息队列传递的是 2 进制数据，不一定非得是文本。
msgsnd 函数用于将数据发送到消息队列。如果该函数被信号打断，会设置 errno 为 EINTR。
参数 msqid：ipc 内核对象 id
参数 msgp：消息数据地址
参数 msgsz：消息正文部分的大小（不包含消息类型）
参数 msgflg：可选项
该值为 0：如果消息队列空间不够，msgsnd 会阻塞。
IPC_NOWAIT：直接返回，如果空间不够，会设置 errno 为 EAGIN.
返回值：0 表示成功，-1 失败并设置 errno。
msgrcv 函数从消息队列取出消息后，并将其从消息队列里删除。
参数 msqid：ipc 内核对象 id
参数 msgp：用来接收消息数据地址
参数 msgsz：消息正文部分的大小（不包含消息类型）
参数 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 分别用于向消息队列发送数据和接收数据。
msg_send 程序定义了一个结构体 Msg，消息正文部分是结构体 Person。该程序向消息队列发送了 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，却无足够的权限执行
问题：
在Linux 系统中通过消息队列进行进程间的通讯时，只要定义的BufSize小于1024，队列就能正常读写，当Size定义大于1024时，队列就无法成功。
处理步骤：
SystemV的消息队列
/etc/sysctl.conf
修改
kernel.msgmni=1000
kernel.msgmax=81920
kernel.msgmnb=163840
msgmni为MSGMNI，即系统的消息队列数目。平台每个DTA需要使用3个消息队列，即最大DTA数为1000/3。该参数应该比平台最大队列个数参数配置大。
msgmax为MSGMAX，即一个消息的字节大小。目前扩展值为8k，平台一个交易消息为4个字节，不会超过限制。
msgmnb为MSGMNB，即队列存放消息的总字节数。
POSIX消息队列
修改
fs.mqueue.msg_max=1000 <-消息个数
fs. mqueue. msgsize_max=8192 <-消息长度
另外 *** 作系统对文件大小的限制ulimit -q你可以看到POSIX消息队列的最大容量
cat /proc/sys/kernel/msgmax
cat /proc/sys/kernel/msgmni
cat /proc/sys/kernel/msgmnb
因为不仅仅信号量，共享内存、消息队列在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>
#include<sys/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=0i<50i++)
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=0i<50i++)
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
}
欢迎分享，转载请注明来源：内存溢出
微信扫一扫
支付宝扫一扫
评论列表（0条）