Linux C++ 实现简单的广播服务

简单的广播服务：就是客户端可以关注某个话题，当某个话题有信息发布了之后，关注了该话题的客户端就可以收到该信息。

服务端主要是通过epoll来处理所有连接的文件描述符，某个文件描述符有事件产生就处理。

客户端有两个，一个是关注话题的，一个是发布话题信息的。

关注话题的客户端通过TCP连接，首先发送关注的话题，再等待话题信息的到来。

发布话题信息的客户端通过UDP将某个话题信息发送到服务端，服务端再转发给关注了该话题的客户端。

下面的Client.h和Client.cpp主要是管理客户关注的话题以及客户连接的文件描述符。

代码中有注释，都可以看懂的。

Client.h

#ifndef _CLIENT_H_
#define _CLIENT_H_

#include <vector>
#include <iostream>
#include <stdio.h>
#include <string>

using std::vector;
using std::string;

//客户信息类
struct Client
{
int fd; //客户连接的sockfd
vector<string> topic; //客户关注的话题
};
//客户信息管理类
class Client_Data
{
private:
vector<Client> client; //客户信息数组
Client_Data(){}

public:
~Client_Data(){}
static Client_Data* getInstance(); //获取Client_Data对象
void AddClient(struct Client cli); //新的客户连接，添加客户信息
void RemClient(int sockfd); //客户关闭连接，删除客户信息
void AddTopic(int fd, vector<string> topic); //添加用户关注的话题
vector<int> FindFdOfTopic(string topic); //根据话题查找哪些客户关注了該话题
void printClient(); //打印Client,测试用
};

#endif

Client.cpp
#include "Client.h"

//获取管理客户信息的类，单例
Client_Data* Client_Data::getInstance()
{
static Client_Data *Cdata = NULL;
if(Cdata == NULL)
Cdata = new Client_Data();
return Cdata;
}
/*
客户信息管理类由主线程调用，不需要锁
*/
//添加客户信息
void Client_Data::AddClient(struct Client cli)
{
client.push_back(cli);
}
//删除客户信息
void Client_Data::RemClient(int sockfd)
{
vector<Client>::iterator it;
for(it = client.begin(); it != client.end(); ++it)
{
if(it->fd == sockfd)
{
client.erase(it);
break;
}
}
}

//向客户添加他所关注的话题
void Client_Data::AddTopic(int fd, vector<string> topic)
{
vector<Client>::iterator it;
for(it = client.begin(); it != client.end(); ++it)
{
if(it->fd == fd)
{
it->topic = topic;
break;
}
}
}

//根据话题获取客户
vector<int> Client_Data::FindFdOfTopic(string topic)
{
vector<int> ans;
vector<Client>::iterator it;
vector<string>::iterator sit;
for(it = client.begin(); it != client.end(); ++it)
{
for(sit = it->topic.begin(); sit != it->topic.end(); ++sit)
{
if(*sit == topic)
{
ans.push_back(it->fd);
}
}
}
return ans;
}

void Client_Data::printClient()
{
vector<Client>::iterator it;
vector<string>::iterator sit;
for(it = client.begin(); it != client.end(); ++it)
{
printf("%d:", it->fd);
for(sit = it->topic.begin(); sit != it->topic.end(); ++sit)
{
printf(" %s", sit->c_str());
}
printf("\n");
}
}


main函数的cpp文件
#include <cstdio>
#include <sys/socket.h>
#include <sys/types.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <unistd.h>
#include <stdlib.h>
#include <errno.h>
#include <cstring>
#include <sys/epoll.h>
#include <fcntl.h>
#include <vector>
#include <string>

#include "../thread_pool.h"
#include "Client.h"

#define MAXBUFFER 1024
#define MAXLINE 512
#define MAXEVENTNUMBER 50

epoll_event events[MAXEVENTNUMBER];
int epollfd;

class task;

threadpool<task> *pool = NULL;
Client_Data *client = NULL;

//任务类
class task
{
private:
int fd; //客户sockfd
string message; //发送的消息
public:
task(int fd, string message);
~task();
void doit();
};
//构造函数
task::task(int f, string mess) : fd(f), message(mess)
{}
//析构函数
task::~task()
{}
//任务执行函数，将消息发送给客户
void task::doit()
{
const char *p = message.c_str();
int ret = write(fd, p, strlen(p));
if(ret < 0)
{
printf("write error: %s\n", strerror(errno));
}
}

//将文件描述符设置成非阻塞模式
int setnonblocking(int fd)
{
int old_option = fcntl(fd, F_GETFL);
int new_option = old_option | O_NONBLOCK;
fcntl(fd, F_SETFL, new_option);
return old_option;
}
//文件描述符fd添加事件EPOLLIN，并注册到epollfd的epoll内核事件表,默认LT模式
void addfd(int epollfd, int fd)
{
epoll_event event;
event.data.fd = fd;
event.events = EPOLLIN;
epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, &event);
setnonblocking(fd);
}

void Epoll(int listenfd, int udpfd)
{
int i, ret, connfd; //连接套接字
socklen_t len;
sockaddr_in cliaddr;//客户地址信息
for( ; ; )
{
//获取已经就绪的描述符
ret = epoll_wait(epollfd, events, MAXEVENTNUMBER, -1);
if(ret < 0)
{//出错处理
printf("epoll_wait error: %s\n", strerror(errno));
return;
}
for(i = 0; i < ret; ++i)
{
int fd = events[i].data.fd;
if(fd == listenfd) //新的连接到来
{
len = sizeof(cliaddr);
connfd = accept(listenfd, (struct sockaddr *)&cliaddr, &len);
char buf[20];
printf("IP %s conn\n", inet_ntop(AF_INET,
&cliaddr.sin_addr, buf, sizeof(buf)));
Client cli;
cli.fd = connfd;
client->AddClient(cli);
addfd(epollfd, connfd);//添加进epollfd的epoll内核事件表
}
else if(fd == udpfd)
{//udp发送推送话题信息的到来
char buffer[MAXBUFFER];
char temp[20], topic[50];
int index = 0, k = 0;
memset(buffer, '\0', MAXBUFFER);
struct sockaddr_in cliaddr;
socklen_t len = sizeof(cliaddr);
//接收推送的话题消息
int ret = recvfrom(udpfd, buffer, MAXBUFFER - 1,
0, (struct sockaddr *)&cliaddr, &len);
if(ret < 0)
{
printf("UDP recvfrom error : %s\n", strerror(errno));
continue;
}
buffer[ret] = '\0';
//查看消息头，判断是否为推送话题消息
//推送消息格式:push#topic#messge
while(buffer[index] != '#' && buffer[index] != '\0')
{
temp[k] = buffer[index];
++k; ++index;
}
++index;
temp[k] = '\0';
if(!strcmp(temp, "push")) //推送话题消息
{
k = 0;
//获取话题
while(buffer[index] != '#' && buffer[index] != '\0')
{
topic[k] = buffer[index];
++k; ++index;
}
++index;
topic[k] = '\0';
string str(topic);
//查找关注了该话题的客户
vector<int> fds = client->FindFdOfTopic(str);
char *p = buffer + index;
//printf("fds %d\n", (int)fds.size());
//client->printClient();
sprintf(buffer, "%s: %s", topic, p); //在消息头添加话题信息
string mess(buffer); //最后部分为推送的消息
//printf("%s\n", mess.c_str());
vector<int>::iterator it;
//向关注了该话题的客户推送消息
for(it = fds.begin(); it != fds.end(); ++it)
{
task *ta = new task(*it, mess);
//printf("%d \n", *it);
pool->append_task(ta);
}
//回复发布话题消息成功
sendto(udpfd, "push success", 12, 0,
(struct sockaddr *)&cliaddr, len);
}
else
{//回复发布话题消息失败
sendto(udpfd, "push error", 10, 0,
(struct sockaddr *)&cliaddr, len);
}
}
else if(events[i].events & EPOLLIN)
{//LT模式，有数据可读就触发
char buffer[MAXBUFFER];

memset(buffer, '\0', sizeof(buffer));
int getByte = read(fd, buffer, MAXBUFFER - 1); //读取消息
buffer[getByte] = '\0';
if(getByte < 0)
{
printf("read buffer error : %s\n", strerror(errno));
continue;
}
else if(getByte == 0) //客户关闭连接。删除客户信息
{
close(fd);
client->RemClient(fd);
break;
}
else
{//关注话题消息格式为: get#topic#topic#...#topic#
char temp[20], topic[50];
int index = 0, k = 0;
//查看消息头，判断是否关注话题
while(buffer[index] != '#' && buffer[index] != '\0')
{
temp[k] = buffer[index];
++k; ++index;
}
++index;
temp[k] = '\0';
if(!strcmp(temp, "get")) //关注话题
{
vector<string> top; //话题的数组
while(buffer[index] != '\0')
{
k = 0;
while(buffer[index] != '#') //获取每个话题
{
topic[k] = buffer[index];
++k; ++index;
}
++index;
topic[k] = '\0';
string s_topic(topic);
top.push_back(s_topic);
}
client->AddTopic(fd, top);//为该客户添加他所关注的话题
string mess("follow success"); //回复成功关注消息
task *ta = new task(fd, mess);
pool->append_task(ta);
}
else //接收消息头部信息不匹配
{
string mess("follow error"); //回复关注失败消息
task *ta = new task(fd, mess);
pool->append_task(ta);
}
}
}
else
{
printf("Other things happened\n");
}
}
}
}

int main(int argc, char *argv[])
{
if(argc != 2)
{
printf("usage: %s port", argv[0]);
return 1;
}
//监听套接字，UDP连接套接字
int listenfd, udpfd;
struct sockaddr_in seraddr, udpaddr;
int port = atoi(argv[1]); //端口

//初始化服务端的sockaddr_in
bzero(&seraddr, sizeof(seraddr));
seraddr.sin_family = AF_INET;
seraddr.sin_port = htons(port);
seraddr.sin_addr.s_addr = htonl(INADDR_ANY);

//创建监听套接字
listenfd = socket(AF_INET, SOCK_STREAM, 0);
if(listenfd < 0)
{
printf("socket error: %s\n", strerror(errno));
return 1;
}
//绑定
int ret = bind(listenfd, (struct sockaddr *)&seraddr, sizeof(seraddr));
if(ret < 0)
{
printf("bind error: %s\n", strerror(errno));
return 1;
}
//监听
listen(listenfd, 10);

//创建udp套接字
bzero(&udpaddr, sizeof(udpaddr));
udpaddr.sin_family = AF_INET;
udpaddr.sin_port = htons(port);
udpaddr.sin_addr.s_addr = htonl(INADDR_ANY);

udpfd = socket(AF_INET, SOCK_DGRAM, 0);
if(udpfd < 0)
{
printf("UDP socket error: %s\n", strerror(errno));
return 1;
}
ret = bind(udpfd, (struct sockaddr *)&udpaddr, sizeof(udpaddr));
if(ret < 0)
{
printf("UDP bind error: %s\n", strerror(errno));
return 1;
}

//创建epoll的文件描述符
epollfd = epoll_create(20);
if(epollfd < 0)
{
printf("epollfd create error: %s\n", strerror(errno));
return 1;
}
addfd(epollfd, listenfd);
addfd(epollfd, udpfd);
//创建客户管理对象
client = Client_Data::getInstance();

//开启线程池，用于推送话题消息。
pool = new threadpool<task>(5, 10);
pool->start();

//epoll
Epoll(listenfd, udpfd);
return 0;
}
关注话题的客户端代码：

很简单，首先发送一个关注的信息，例如：get#topic#topic#...#topic#

#include <stdio.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <arpa/inet.h>
#include <errno.h>
#include <netinet/in.h>
#include <string.h>
#include <unistd.h>
#include <stdlib.h>

#define MAXBUFFER 2048
#define MAXLINE 1024

int main(int argc, char *argv[])
{
if(argc < 4)
{
printf("usage: %s IP port topic...\n", argv[0]);
return 1;
}
char buffer[MAXBUFFER];
char message[MAXLINE];
int sockfd, i;
size_t size;
struct sockaddr_in seraddr;
int port = atoi(argv[2]);

bzero(&seraddr, sizeof(seraddr));
seraddr.sin_family = AF_INET;
seraddr.sin_port = htons(port);
inet_pton(AF_INET, argv[1], &seraddr.sin_addr);

sockfd = socket(AF_INET, SOCK_STREAM, 0);
if(sockfd < 0)
{
printf("socket error: %s\n", strerror(errno));
return 1;
}

connect(sockfd, (struct sockaddr *)&seraddr, sizeof(seraddr));

//获取关注话题的消息
sprintf(buffer, "get#");
for(i = 3; i < argc; ++i)
{
sprintf(buffer, "%s%s#", buffer, argv[i]);
}

write(sockfd, buffer, strlen(buffer));
//读取关注是否成功
size = read(sockfd, message, MAXLINE - 1);
printf("%s\n", message);
//等待话题消息
while(1)
{
size = read(sockfd, message, MAXLINE - 1);
message[size] = '\0';
if(size == 0)
{
printf("server close\n");
break;
}
else if(size < 0)
{
printf("read error: %s\n", strerror(errno));
continue;
}
printf("%s\n", message);
}
return 0;
}



推送话题信息的代码，很简单的
#include <stdio.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <sys/types.h>
#include <string.h>
#include <unistd.h>
#include <netinet/in.h>
#include <stdlib.h>

#define MAXLINE 1024
#define MAXBUFFER 2048

int main(int argc, char *argv[])
{
if(argc != 3)
{
printf("usage: %s IP port\n", argv[0]);
return 1;
}
char topic[50];
char message[MAXLINE];
char buffer[MAXBUFFER];
int sockfd;
int port = atoi(argv[2]);
struct sockaddr_in seraddr;

bzero(&seraddr, sizeof(seraddr));
seraddr.sin_family = AF_INET;
seraddr.sin_port = htons(port);
inet_pton(AF_INET, argv[1], &seraddr.sin_addr);

sockfd = socket(AF_INET, SOCK_DGRAM, 0);
printf("Input the topic: ");
scanf("%s", topic);
printf("Input the message: ");
getchar();
scanf("%[^\n]", message);

sprintf(buffer, "push#%s#%s", topic, message);
printf("buffer:%s\n", buffer);

sendto(sockfd, buffer, strlen(buffer), 0, (struct sockaddr *)&seraddr,
sizeof(seraddr));

recvfrom(sockfd, message, MAXLINE, 0, NULL, NULL);
printf("message:%s\n", message);
close(sockfd);
return 0;
}

很简单的广播服务，主要是Socket编程。一起学习，一起成长。