微言Netty：百万并发基石上的epoll之剑

 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

/**
 * @author shichaoyang
 * @Description: 数据同步器
 * @date 2020-10-14 19:41
 */
public class DataSynchronizer {
    /**
     * 加载本地底层C实现库
     */
    static {
        System.loadLibrary("synchronizer");
    }
    /**
     * 底层数据同步方法
     */
    private native String syncData(String status);
    /**
     * 程序启动，调用底层数据同步方法
     *
     * @param args
     */
    public static void main(String... args) {
        String rst = new DataSynchronizer().syncData("ProcessStep2");
        System.out.println("The execute result from C is : " + rst);
    }
}

1
2
3
4
5
6
7

执行apt install default-jdk 安装java环境，安装完毕。

通过update-alternatives --list java 获取java安装路径，这里为：/usr/lib/jvm/java-11-openjdk-amd64   

设置java环境变量 export JAVA_HOME=/usr/lib/jvm/java-11-openjdk-amd64

环境设置完毕之后，就可以开始进行下一步了。

1

javac -h . DataSynchronizer.java

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

/* DO NOT EDIT THIS FILE - it is machine generated */
#include <jni.h>
/* Header for class DataSynchronizer */
#ifndef _Included_DataSynchronizer
#define _Included_DataSynchronizer
#ifdef __cplusplus
extern "C" {
#endif
/*
 * Class:     DataSynchronizer
 * Method:    syncData
 * Signature: (Ljava/lang/String;)Ljava/lang/String;
 */
JNIEXPORT jstring JNICALL Java_DataSynchronizer_syncData
  (JNIEnv *, jobject, jstring);
#ifdef __cplusplus
}
#endif
#endif

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

#include <jni.h>
#include <stdio.h>
#include "DataSynchronizer.h"
 
JNIEXPORT jstring JNICALL Java_DataSynchronizer_syncData(JNIEnv *env, jobject obj, jstring str) {
   // Step 1: Convert the JNI String (jstring) into C-String (char*)
   const char *inCStr = (*env)->GetStringUTFChars(env, str, NULL);
   if (NULL == inCStr) {
        return NULL;
    }
 
   // Step 2: Perform its intended operations
   printf("In C, the received string is: %s\n", inCStr);
   (*env)->ReleaseStringUTFChars(env, str, inCStr);  // release resources
 
   // Prompt user for a C-string
   char outCStr[128];
   printf("Enter a String: ");
   scanf("%s", outCStr);
 
   // Step 3: Convert the C-string (char*) into JNI String (jstring) and return
   return (*env)->NewStringUTF(env, outCStr);
}

1

gcc -fPIC -I"$JAVA_HOME/include" -I"$JAVA_HOME/include/linux" -shared -o libsynchronizer.so DataSynchronizer.c

1

java -Djava.library.path=. DataSynchronizer

1
2
3
4

java -Djava.library.path=. DataSynchronizer
In C, the received string is: ProcessStep2
Enter a String: sdfsdf
The execute result from C is : sdfsdf

1

int select(int nfds, fd_set *readfds, fd_set *writefds, fd_set *errorfds, struct timeval *timeout);

 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

//创建server端套接字，获取文件描述符
    int listenfd = socket(PF_INET,SOCK_STREAM,0);
    if(listenfd < 0) return -1;
    
    //绑定服务器
    bind(listenfd,(struct sockaddr*)&address,sizeof(address));
    //监听服务器
    listen(listenfd,5); 
    
    struct sockaddr_in client;
    socklen_t addr_len = sizeof(client);
    
    //接收客户端连接
    int connfd = accept(listenfd,(struct sockaddr*)&client,&addr_len);
    
    //读缓冲区
    char buff[1024]; 
    
    //读文件操作符
    fd_set read_fds;  
    
    while(1)
    {
        memset(buff,0,sizeof(buff));
        
        //注意：每次调用select之前都要重新设置文件描述符connfd，因为文件描述符表会在内核中被修改
        FD_ZERO(&read_fds);
        FD_SET(connfd,&read_fds);
        
        //注意：select会将用户态中的文件描述符表放到内核中进行修改，内核修改完毕后再返回给用户态，开销较大
        ret = select(connfd+1,&read_fds,NULL,NULL,NULL);
        if(ret < 0)
        {
            printf("Fail to select!\n");
            return -1;
        }
        
        //检测文件描述符表中相关请求是否可读
        if(FD_ISSET(connfd, &read_fds))
        {
            ret = recv(connfd,buff,sizeof(buff)-1,0);
            printf("receive %d bytes from client: %s \n",ret,buff);
        }
    }

1

int poll(struct pollfd *fds, nfds_t nfds, int timeout);

1
2
3
4
5

struct pollfd {
    int   fd;         /* 文件描述符 */
    short events;     /* 关心的事件 */
    short revents;    /* 实际返回的事件 */
};

 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

 //创建server端套接字，获取文件描述符
    int listenfd = socket(PF_INET,SOCK_STREAM,0);
    if(listenfd < 0) return -1;
    
    //绑定服务器
    bind(listenfd,(struct sockaddr*)&address,sizeof(address));
    //监听服务器
    listen(listenfd,5); 
    
    struct pollfd pollfds[1];
    socklen_t addr_len = sizeof(client);
    
    //接收客户端连接
    int connfd = accept(listenfd,(struct sockaddr*)&client,&addr_len);
    
    //放入fd数组
    pollfds[0].fd = connfd;
    pollfds[0].events = POLLIN;
    
    //读缓冲区
    char buff[1024]; 
    
    //读文件操作符
    fd_set read_fds;  
    
    while(1)
    {
        memset(buff,0,sizeof(buff));
        
        /**
         ** SELECT模型专用
         ** 注意：每次调用select之前都要重新设置文件描述符connfd，因为文件描述符表会在内核中被修改
         ** FD_ZERO(&read_fds);
         ** FD_SET(connfd,&read_fds);
        ** 注意：select会将用户态中的文件描述符表放到内核中进行修改，内核修改完毕后再返回给用户态，开销较大
        ** ret = select(connfd+1,&read_fds,NULL,NULL,NULL);
        **/
        
        ret = poll(pollfds, 1, 1000);
        if(ret < 0)
        {
            printf("Fail to poll!\n");
            return -1;
        }
        
        /**
         ** SELECT模型专用
         ** 检测文件描述符表中相关请求是否可读
         ** if(FD_ISSET(connfd, &read_fds))
         ** {
         **   ret = recv(connfd,buff,sizeof(buff)-1,0);
         **   printf("receive %d bytes from client: %s \n",ret,buff);
         ** }
         **/
        //检测文件描述符数组中相关请求
        if(pollfds[0].revents & POLLIN){
            pollfds[0].revents = 0;
            ret = recv(connfd,buff,sizeof(buff)-1,0);
            printf("receive %d bytes from client: %s \n",ret,buff);
        }
    }

1
2
3

-->ET Mode: it was triggered once
get 1 bytes of content: A
-->wait to read!

1
2
3
4

-->ET Mode: it was triggered once
get 9 bytes of content: ABCDEFGHI
get 8 bytes of content: JKLMNOPQ
-->wait to read!

1
2

-->LT Mode: it was triggered once!
get 1 bytes of content: A

1
2
3
4

-->LT Mode: it was triggered once!
get 9 bytes of content: ABCDEFGHI
-->LT Mode: it was triggered once!
get 8 bytes of content: JKLMNOPQ

  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
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187

#include <stdio.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <string.h>
#include <fcntl.h>
#include <stdlib.h>
#include <sys/epoll.h>
#include <pthread.h>
#include <errno.h>
#include <stdbool.h>
#define MAX_EVENT_NUMBER 1024   //事件总数量
#define BUFFER_SIZE 10          //缓冲区大小，这里为10个字节
#define ENABLE_ET 0             //ET模式
/* 文件描述符设为非阻塞状态
 * 注意：这个设置很重要，否则体现不出高性能
 */
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放入到内核中的epoll数据结构中并将fd设置为EPOLLIN可读，同时根据ET开关来决定使用水平触发还是边缘触发模式 
 * 注意：默认为水平触发，或上EPOLLET则为边缘触发
*/
void AddFd(int epoll_fd, int fd, bool enable_et)
{
    struct epoll_event event;  //为当前fd设置事件
    event.data.fd = fd;        //指向当前fd
    event.events = EPOLLIN;    //使得fd可读
    if(enable_et)
    {
        event.events |= EPOLLET; //设置为边缘触发
    }
    epoll_ctl(epoll_fd, EPOLL_CTL_ADD, fd, &event);  //将fd添加到内核中的epoll实例中
    SetNonblocking(fd);  //设为非阻塞模式                      
}
/*  LT水平触发 
 *  注意：水平触发简单易用，性能不高，适合低并发场合
 *        一旦缓冲区有数据，则会重复不停的进行通知，直至缓冲区数据读写完毕
 */
void lt_process(struct epoll_event* events, int number, int epoll_fd, int listen_fd)
{
    char buf[BUFFER_SIZE];
    int i;
    for(i = 0; i < number; i++) //已经就绪的事件，这些时间可读或者可写
    {
        int sockfd = events[i].data.fd; //获取描述符
        if(sockfd == listen_fd)  //如果监听类型的描述符，则代表有新的client接入，则将其添加到内核中的epoll结构中
        {
            struct sockaddr_in client_address;
            socklen_t client_addrlength = sizeof(client_address);
            int connfd = accept(listen_fd, (struct sockaddr*)&client_address, &client_addrlength); //创建连接并返回文件描述符（实际进行的三次握手过程）
            AddFd(epoll_fd, connfd, false);  //添加到epoll结构中并初始化为LT模式
        }
        else if(events[i].events & EPOLLIN) //如果客户端有数据过来
        {
            printf("-->LT Mode: it was triggered once!\n");
            memset(buf, 0, BUFFER_SIZE); 
            int ret = recv(sockfd, buf, BUFFER_SIZE - 1, 0);
            if(ret <= 0)  //读取数据完毕后，关闭当前描述符
            {
                close(sockfd);
                continue;
            }
            printf("get %d bytes of content: %s\n", ret, buf);
        }
        else
        {
            printf("something unexpected happened!\n");
        }
    }
}
/*  ET Work mode features: efficient but potentially dangerous */
/*  ET边缘触发
 *  注意：边缘触发由于内核不会频繁通知，所以高效，适合高并发场合，但是处理不当将会导致严重事故
          其通知机制和触发方式参见之前讲解，由于不会重复触发，所以需要处理好缓冲区中的数据，避免脏读脏写或者数据丢失等
 */
void et_process(struct epoll_event* events, int number, int epoll_fd, int listen_fd)
{
    char buf[BUFFER_SIZE];
    int i;
    for(i = 0; i < number; i++)
    {
        int sockfd = events[i].data.fd;
        if(sockfd == listen_fd) //如果有新客户端请求过来，将其添加到内核中的epoll结构中并默认置为ET模式
        {
            struct sockaddr_in client_address;
            socklen_t client_addrlength = sizeof(client_address);
            int connfd = accept(listen_fd, (struct sockaddr*)&client_address, &client_addrlength);
            AddFd(epoll_fd, connfd, true); 
        }
        else if(events[i].events & EPOLLIN) //如果客户端有数据过来
        {
            printf("-->ET Mode: it was triggered once\n");
            while(1) //循环等待
            {
                memset(buf, 0, BUFFER_SIZE);
                int ret = recv(sockfd, buf, BUFFER_SIZE - 1, 0);
                if(ret < 0)
                {
                    if(errno == EAGAIN || errno == EWOULDBLOCK) //通过EAGAIN检测，确认数据读取完毕
                    {
                        printf("-->wait to read!\n");
                        break;
                    }
                    close(sockfd);
                    break;
                }
                else if(ret == 0) //数据读取完毕，关闭描述符
                {
                    close(sockfd);
                }
                else //数据未读取完毕，继续读取
                {
                    printf("get %d bytes of content: %s\n", ret, buf);
                }
            }
        }
        else
        {
            printf("something unexpected happened!\n");
        }
    }
}
int main(int argc, char* argv[])
{
    const char* ip = "10.0.76.135";
    int port = 9999;
    
    //套接字设置这块，参见https://www.gta.ufrj.br/ensino/eel878/sockets/sockaddr_inman.html
    int ret = -1;
    struct sockaddr_in address; 
    bzero(&address, sizeof(address));
    address.sin_family = AF_INET;
    inet_pton(AF_INET, ip, &address.sin_addr);
    address.sin_port = htons(port);
    int listen_fd = socket(PF_INET, SOCK_STREAM, 0);    //创建套接字并返回描述符
    if(listen_fd < 0)
    {
        printf("fail to create socket!\n");
        return -1;
    }
    ret = bind(listen_fd, (struct sockaddr*)&address, sizeof(address)); //绑定本机
    if(ret == -1)
    {
        printf("fail to bind socket!\n");
        return -1;
    }
    ret = listen(listen_fd, 5); //在端口上监听
    if(ret == -1)
    {
        printf("fail to listen socket!\n");
        return -1;
    }
    struct epoll_event events[MAX_EVENT_NUMBER];
    int epoll_fd = epoll_create(5);  //在内核中创建epoll实例，flag为5只是为了分配空间用，实际可以不用带
    if(epoll_fd == -1)
    {
        printf("fail to create epoll!\n");
        return -1;
    }
    AddFd(epoll_fd, listen_fd, true); //添加文件描述符到epoll对象中
    while(1)
    {
        int ret = epoll_wait(epoll_fd, events, MAX_EVENT_NUMBER, -1); //拿出就绪的文件描述符并进行处理
        if(ret < 0)
        {
            printf("epoll failure!\n");
            break;
        }
        if(ENABLE_ET) //ET处理方式
        {
            et_process(events, ret, epoll_fd, listen_fd);
        }
        else  //LT处理方式
        {
            lt_process(events, ret, epoll_fd, listen_fd);
        }
    }
    close(listen_fd); //退出监听
    return 0;
}