【Linux学习笔记】消息队列

方法一：

消息队列的接口

#include "fcntl.h"
#include "sys/stat.h"
#include "mqueue.h"

// 创建消息队列实例。name: 消息队列名称。成功返回0，失败返回-1，错误码存于error中
mqd_t mq_open(const char *name, int oflag,  mode_t mode, struct mq_attr *attr);

// 无限阻塞方式接收消息。成功返回0，失败返回-1
mqd_t mq_receive(mqd_t mqdes, char *msg_ptr, size_t msg_len, unsigned *msg_prio);

// 指定超时时间阻塞方式接收消息。成功返回0，失败返回-1
mqd_t mq_timedreceive(mqd_t mqdes, char *msg_ptr, size_t msg_len, unsigned *msg_prio, const struct timespec *abs_timeout);

// 无限阻塞方式发送消息。成功返回0，失败返回-1
mqd_t mq_send(mqd_t mqdes, const char *msg_ptr, size_t msg_len, unsigned msg_prio);

// 指定超时时间阻塞方式发送消息。成功返回0，失败返回-1
mqd_t mq_timedsend(mqd_t mqdes, const char *msg_ptr, size_t msg_len, unsigned msg_prio, const struct timespec *abs_timeout);

// 关闭消息队列。 成功返回0，失败返回-1
mqd_t mq_close(mqd_t mqdes);

// 分离消息队列。 成功返回0，失败返回-1
mqd_t mq_unlink(const char *name);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

sender.cpp

#include   // For O_* constants
#include 
#include 
#include 
#include 
#include   // For mode constants
#include 

#define MQ_MSG_MAX_SIZE 512  // 最大消息长度
#define MQ_MSG_MAX_ITEM 5    // 最大消息数目

static mqd_t s_mq;

typedef struct _msg_data
{
    char buf[128];
    int num;
}msg_data_t;


void send_data(void)
{
    static int num = 0;
    msg_data_t send_data = {0};

    ++num;
    strcpy(send_data.buf, "Hello Randy");
    send_data.num = num;
    int ret = mq_send(s_mq, (char*)&send_data, sizeof(send_data), 0);
    if (ret < 0)
    {
        perror("mq_send error");
        return;
    }
    printf("send msg = %s, num = %d\n", send_data.buf, send_data.num);
}

int main(void) {
    int ret = 0;
    struct mq_attr attr;

    // 创建消息队列
    memset(&attr, 0, sizeof(attr));
    attr.mq_maxmsg = MQ_MSG_MAX_ITEM;
    attr.mq_msgsize = MQ_MSG_MAX_SIZE;
    attr.mq_flags = 0;
    s_mq = mq_open("/randy1", O_CREAT | O_RDWR, 0777, &attr);
    if (-1 == s_mq) {
        perror("mq_open error");
        return -1;
    }

    for (size_t i = 0; i < 10; i++) {
        send_data();
        sleep(1);
    }

    mq_close(s_mq);

    return 0;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

receiver.cpp

#include   // For O_* constants
#include 
#include 
#include 
#include 
#include   // For mode constants
#include 

#define MQ_MSG_MAX_SIZE 512  // 最大消息长度
#define MQ_MSG_MAX_ITEM 5    // 最大消息数目

static mqd_t s_mq;

typedef struct _msg_data
{
    char buf[128];
    int num;
}msg_data_t;


void receive_data(void)
{
    static int num = 0;
    msg_data_t receive_data = {0};
    int ret = mq_receive(s_mq, (char*)&receive_data, MQ_MSG_MAX_SIZE, 0);
    if (ret < 0)
    {
        perror("mq_send error");
        return;
    }
    printf("send msg = %s, num = %d\n", receive_data.buf, receive_data.num);
}

int main(void) {
    int ret = 0;
    struct mq_attr attr;

    // 创建消息队列
    memset(&attr, 0, sizeof(attr));
    attr.mq_maxmsg = MQ_MSG_MAX_ITEM;
    attr.mq_msgsize = MQ_MSG_MAX_SIZE;
    attr.mq_flags = 0;
    s_mq = mq_open("/randy1", O_RDWR, 0777, &attr);
    if (-1 == s_mq) {
        perror("mq_open error");
        return -1;
    }

    for (size_t i = 0; i < 10; i++) {
        receive_data();
        sleep(1);
    }

    mq_close(s_mq);

    return 0;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

注意⚠️ man 3 mq_receive 可以知道，接受指定的 msg_len 必须大于等于 attr 中的mq_msgsize；同理，发送时，必须小于attr指定的mq_msgsize

方法二

#include   
#include   
#include   
#include   
#include   
#include   
#include   
#include 
#include 



struct my_msg_st {  
    long int my_msg_type;  
    char text[100];  
};

int send_msg(int msqid, const std::string &text) {  
    my_msg_st msg;  
    msg.my_msg_type = 1; // 消息类型  
    strncpy(msg.text, text.c_str(), sizeof(msg.text)); // 复制文本到消息结构体  
  
    if (msgsnd(msqid, &msg, sizeof(msg), 0) == -1) { // 发送消息  
        perror("msgsnd");  
        return -1;  
    }  
    return 0;  
}

int receive_msg(int msqid) {  
    my_msg_st msg;  
    if (msgrcv(msqid, &msg, sizeof(msg), 1, 0) == -1) { // 接收消息  
        perror("msgrcv");   
        return -1;   
    }  
    std::cout << "Received: " << msg.text << std::endl; // 打印接收到的消息  
    return 0;  
}

int main() {  
    int msqid;  
    key_t key = ftok("/tmp", 'R'); // 使用文件系统路径作为关键字，创建唯一的键值  
    if ((msqid = msgget(key, IPC_CREAT | 0666)) == -1) { // 创建消息队列  
        perror("msgget");  
        return -1;  
    }  
    std::cout << "Message queue created with id: " << msqid << std::endl;  
    send_msg(msqid, "Hello, World!"); // 发送消息  
    receive_msg(msqid); // 接收消息  
    return 0;  
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51