메세지 큐 (Message Queue)

오늘은 메세지 큐 구현에 대해 설명을 하고자 합니다.

 

메세지 큐는 IPC 통신 방법 (파이프, 메세지 큐, 공유 메모리 등) 중 하나입니다.

메세지 큐 역시 단방향 통신이며, 전송되는 자료는

큐의 크기에 따라 상대방이 가져가지 않는다고 하더라도

큐 크기만큼 계속 전송이 가능합니다.

반대로 수신자는 전송자가 전송하지 않더라도

큐에 있는 자료를 가져갈 수 있습니다.

예를 들어, 송신자가 30번째 데이터까지 전송하고 죽었다고 가정하면,

수신자는 30번째 데이터까지는 큐에 있기때문에

송신자의 연결 유무와 관계 없이 받을 수 있다는 얘기입니다.

즉, 데이터가 중간에 소실되거나 순서가 뒤바뀌는 일은 없을 것입니다.

그리고 데이터 타입을 정해서 원하는 데이터만 보내는 것도 가능합니다.

 

type struct {
	long data_type;
	char data_buf[BUF_SIZE];
	...
}MQUEUE;

여기서 중요한 점은 long data_type은 항상 첫번째 위치에 있어야한다는 제약이 있습니다.

 

int msgget(key_t key, int msgflag)

msgflag	mean
IPC_CREAT	키에 해당하는 큐가 있다면 큐의 식별자를 반환하고, 그렇지 않다면 생성함
IPC_EXCL	키에 해당하는 큐가 없다면 생성하고 있다면 -1을 반환하고 복귀함

큐에 대한 번호를 알고 있다면 프로세스끼리 같은 번호의 규를 사용할 수 있습니다.

 

int msgsnd(int msqid, const void* msgp, size_t msgsz, int msgflag)

msgflag	mean
0	큐에 공간이 생길 때까지 기다림
IPC_NOWAIT	큐에 여유 공간이 없다면 바로 -1을 반환하고 복귀함

데이터를 전송할 때 long data_type이 같이 전송되기 때문에 실제 데이터 크기는 long 크기를 제외한 크기입니다.

sizeof(struct) - sizeof(long)

 

메세지 큐 (Message Queue) 송신자

//sender.c
/**
* os : ubuntu 18.04
* ide : eclipse
*/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>

#define BUFSIZE 1024 //buffer size definition

struct dataset{
	long data_type;
	char data_buf[BUFSIZE];
	int data_num;
};

int main()
{
	key_t key = 9999; // queue key value
	int msgq_id;
	int index = 0;
	struct dataset data;

	if((msgq_id = msgget(key, IPC_CREAT | 0666)) == -1)
	{
		perror("msgget() failed!");
		exit(1);
	}

	while(1)
	{
		data.data_type = 1;
		data.data_num = index;
		sprintf(data.data_buf, "index = %d, type = %d, data = %s", data.data_num, data.data_type, data.data_buf);

		if(msgsnd(msgq_id, &data, sizeof(struct dataset) - sizeof(long), 0)) == -1)
		{
			perror("msgsnd() failed");
			exit(1);
		}
	}
    
    return 0;
}

 

ssize_t msgrcv(int msqid, const void* msgp, size_t msgsz, long msgtype, int msgflag)

msgtype	mean
0	큐에 자료가 있다면 첫 번째 자료를 읽음
양	양수로 지정한 값과 같은 데이터를 읽음
음	음수의 절대 값과 같거나 보다 제일 작은 값을 읽음

msgflag	mean
IPC_NOWAIT	메세지 큐에 메세지가 없다면 기다리지 않고 -1 반환하고 복귀
MSG_NOERROR	메세지 큐에 있는 자료가 준비된 데이터 크기보다 크면 초과 부분을 잘라 내고 읽음
0	큐에 데이터가 있다면 읽어드리고, 없다면 기다림

 

메세지 큐 (Message Queue) 수신자

//receiver.c
/**
* os : ubuntu 18.04
* ide : eclipse
*/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>

#define BUFSIZE 1024 //buffer size definition

struct dataset{
long data_type;
char data_buf[BUFSIZE];
int data_num;
};

int main()
{
	key_t key = 9999;
	int msgq_id;
	struct dataset data;

	msgq_id = msgget(key, IPC_CREAT | 0666);
	if(msgq_id < 0)
	{
		perror("msgget error");
		exit(1);
	}
	while(1)
	{
		if(msgrcv(msgq_id, &data, sizeof(struct dataset) - sizeof(long), 0, 0) == -1)
		{
			perror("msgrcv error : ");
			exit(1);
		}
		printf("recv data index : %d, type : %d, data : %s", data.data_num, data.data_type, data_buf);
		sleep(1); 
	}

	return 0;
}

 

구현 결과 테스트