大橙子网站建设,新征程启航

为企业提供网站建设、域名注册、服务器等服务

linux消息队列命令 linux 消息队列 系统参数

Linux 消息队列长度处理

问题:

创新互联公司是专业的黑山网站建设公司,黑山接单;提供网站设计、成都网站制作,网页设计,网站设计,建网站,PHP网站建设等专业做网站服务;采用PHP框架,可快速的进行黑山网站开发网页制作和功能扩展;专业做搜索引擎喜爱的网站,专业的做网站团队,希望更多企业前来合作!

在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

linux消息队列的收发

26203就是你这个程序运行时候的 PID 阿,你在发消息之前,不是把当前进程的PID存放到消息里的 mtype 了吗

p-mtype = getpid();

然后接收方收到这个消息,又把 mtype 打印出来了阿

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

这个就是打印从 PID 收到了消息,消息是 p-mdata里的内容。

Linux系统编程—消息队列

消息队列本质上是位于内核空间的链表,链表的每个节点都是一条消息。每一条消息都有自己的消息类型,消息类型用整数来表示,而且必须大于 0。每种类型的消息都被对应的链表所维护:

其中数字 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,却无足够的权限执行

自己实现消息队列msg queue linux C

因为不仅仅信号量,共享内存、消息队列在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消息队列命令 linux 消息队列 系统参数
分享链接:http://dzwzjz.com/article/doshpoo.html
在线咨询
服务热线
服务热线:028-86922220
TOP