tcp client如何实现重连？

发布于 2021-06-30 16:44:22 浏览：4659 订阅该版

使用的是rtthread+stm32f429，想实现can和tcp的相互转发。
目前有2个线程，tcp线程，使用select方法非阻塞接收tcp消息，收到后调用can发送函数发送出去。can线程，can接收后调用send函数通过tcp client发送出去。
现在测试能正常工作。但是觉得万一通讯有问题，最好能实现重连。

于是在tcp client线程里写了里外两个while(1)循环。外层循环了，调用的是socket(),connect()函数，内层循环里是调用select()根据结果来决定是否接收，出错后关闭socket，然后跳到外层循环。

想的是这样的，可实际操作后，上电通讯成功后，关闭tcp server然后再打开，发现打印出重新重建了socket和重新连接了。电脑端通过can发送信息后，设备接收后也可以通过tcp发送到tcp server.多次发送也没问题。
但电脑端第一次发送tcp消息后，设备接收后可以通过can发过来。电脑端再次发送，接报错了...

```c
while(1)
{
    ...
    sock=-1;
    while(sock==-1)
    {
        sock=socket(AF_INET,SOCK_STREAM,0);
    }
    rt_kprintf("socke is created.\r\n");
    ...
    if (connect(sock,(struct sockaddr*)&server_addr,sizeof(struct    sockaddr))==-1)
    {
        rt_kprintf("Connect fail!\n");
        closesocket(sock);
        rt_free(tcp_recv_data);
        return;
    }
    else
    {
        rt_kprintf("Connect successful\n");
    }
    ...
    while (1)
    {
        FD_ZERO(&readset);
        FD_SET(sock,&readset);
        i=select(maxfdp1,&readset,0,0,&tim);
        if(i==0)
            continue;
        if(FD_ISSET(sock,&readset))
        {
            bytes_received=recv(sock,tcp_recv_data,BUFSZ-1,MSG_DONTWAIT);
            if (bytes_received<0)
            {
                closesocket(sock);
                rt_kprintf("\nreceived  error-1,  close   the socket.\r\n");
                rt_free(tcp_recv_data);
                break;
            }
            else    if(bytes_received==0)
            {
                closesocket(sock);
                rt_kprintf("\nreceived  error0,  close   the socket.\r\n");
                rt_free(tcp_recv_data);
                break;
            }
            ...
        }
    }

```

```
 \ | /
- RT -     Thread Operating System
 / | \     4.0.3 build Jun 26 2021
 2006 - 2020 Copyright by rt-thread team
lwIP-2.0.2 initialized!
[I/sal.skt] Socket Abstraction Layer initialize success.
msh />ifconfig
network interface device: e0 (Default)
MTU: 1500
MAC: 00 80 e1 05 37 2b
FLAGS: UP LINK_UP INTERNET_DOWN DHCP_ENABLE ETHARP BROADCAST IGMP
ip address: 0.0.0.0
gw address: 0.0.0.0
net mask  : 0.0.0.0
dns server #0: 192.168.1.1
dns server #1: 0.0.0.0
msh />can2tcp
socke is created.
Connect successful
tcp_thread: recv data = aabbccdd
tcp_thread: recv data = aabbccdd
tcp_thread: recv data = aabbccdd
ID:1 0 0 0 0 0 0 0 0
can_rx_thread: send message -
ID:1 0 0 0 0 0 0 0 0
can_rx_thread: send message -
ID:1 0 0 0 0 0 0 0 0
can_rx_thread: send message -
ID:1 0 0 0 0 0 0 0 0
can_rx_thread: send message -

received  error0,  close   the socket.
socke is created.
Connect successful
tcp_thread: recv data = aabbccdd
(rt_object_get_type(&mb->parent.parent) == RT_Object_Class_MailBox) assertion failed at function:rt_mb_send_wait, line number:
```

小小李sunny 2021-07-04

这家伙很懒，什么也没写！

4 个回答

cxhxy12345 2021-07-01

这家伙很懒，什么也没写！

Any 2021-07-01

这家伙很懒，什么也没写！

Any 2021-07-06

这家伙很懒，什么也没写！

Any 2021-07-08

这家伙很懒，什么也没写！

请登录后再发布评论，点击登录

追加回复

ZCmengyang 2021-07-01

这家伙很懒，什么也没写！

Any 2021-07-01

这家伙很懒，什么也没写！

ZCmengyang 2021-07-01

这家伙很懒，什么也没写！

Any 2021-07-01

这家伙很懒，什么也没写！

Any 2021-07-09

这家伙很懒，什么也没写！

ZCmengyang 2021-07-09

这家伙很懒，什么也没写！

请登录后再发布评论，点击登录

追加回复

Any 2021-07-03

这家伙很懒，什么也没写！

群里一位好心的朋友帮我改了程序，可以实现tcp重连。但是有个问题，上电后一开始tcp接收不能正常工作，只有tcp重连后，才正常。
我也没明白我的程序和这个程序区别在哪里...

这是我的程序
```c
#include <rtthread.h>
#include "rtdevice.h"

#include <lwip/api.h>
#include <lwip/sockets.h>
#include <lwip/netdb.h>

// #include    <sys/select.h>
// #include    <sys/socket.h>
// #include    <netdb.h>
#include    <string.h>
#include    <finsh.h>

#define BUFSZ   1024

#define CAN_DEV_NAME    "can1"

static  int sock;

ALIGN(RT_ALIGN_SIZE)
static char thread1_stack[1024];
static struct rt_thread thread1;

static  struct rt_semaphore rx_sem;
static  rt_device_t can_dev;

static  rt_err_t    can_rx_call(rt_device_t dev,    rt_size_t   size)
{
    rt_sem_release(&rx_sem);
    return  RT_EOK;
}

static  void    can_rx_thread(void  *parameter)
{
    int i;
    rt_err_t    res;
    struct rt_can_msg   rxmsg={0};
    char    tcp_send_data[]="hahaha\r\n";
    
    rt_device_set_rx_indicate(can_dev,  can_rx_call);
    while(1)
    {
        rt_sem_take(&rx_sem,RT_WAITING_FOREVER);
        rt_device_read(can_dev,0,&rxmsg,sizeof(rxmsg));
        rt_kprintf("ID:%x",rxmsg.id);
        for(i=0;i<8;i++)
        {
            rt_kprintf("%2x",rxmsg.data[i]);
        }
        rt_kprintf("\n");

res = send(sock, rxmsg.data, rxmsg.len, 0);

rt_kprintf("can_rx_thread: send message - %c\n", rxmsg.data[0]);

/* 延时 50ms */
        rt_thread_mdelay(50);
    }
}

/* 线程 1 入口函数 */
void can2tcp(int  argc,char *argv[])
{
    int i;
    struct rt_can_msg can_msg = {0};
    rt_err_t    res;
    rt_size_t   can_size;
    rt_thread_t thread_can_rx;
    char    can_name[RT_NAME_MAX];

char    *tcp_recv_data;
    struct  hostent *host;
    // int sock,   bytes_received;
    int bytes_received;
    struct sockaddr_in  server_addr;
    const   char    *url;
    int port;
    int maxfdp1;
    fd_set readset;
    struct timeval   tim;

uint8_t can_recv_data[8];
    char sendData[]="hahahaha";

url =  "192.168.1.143";
    port   =    5000;

/*-------------初始化tcp client---------------------*/

host    =   gethostbyname(url);
    tcp_recv_data    =   rt_malloc(BUFSZ);
    if(tcp_recv_data==RT_NULL)
    {
        rt_kprintf("No  memory\n");
        return;
    }
    
    // if((sock=socket(AF_INET,SOCK_STREAM,0))==-1)
    // {
    //     rt_kprintf("Socket  error\n");
    //     rt_free(tcp_recv_data);
    //     return;
    // }

// server_addr.sin_family=AF_INET;
    // server_addr.sin_port=htons(port);
    // server_addr.sin_addr=*((struct  in_addr*)host->h_addr);
    // rt_memset(&(server_addr.sin_zero),0,sizeof(server_addr.sin_zero));

// if (connect(sock,(struct sockaddr*)&server_addr,sizeof(struct    sockaddr))==-1)
    // {
    //     rt_kprintf("Connect fail!\n");
    //     closesocket(sock);
    //     rt_free(tcp_recv_data);
    //     return;
    // }
    // else
    // {
    //     rt_kprintf("Connect successful\n");
    // }

/*-------------打开can线程----------------*/

rt_strncpy(can_name,CAN_DEV_NAME,RT_NAME_MAX);

can_dev =   rt_device_find(can_name);
    if(!can_dev)
    {
        rt_kprintf("find    %s  failed!\n",can_name);
        return;
    }

rt_sem_init(&rx_sem,"rx_sem",0,RT_IPC_FLAG_FIFO);

can_msg.id=0x78;
    can_msg.ide=RT_CAN_STDID;
    can_msg.rtr=RT_CAN_DTR;
    can_msg.len=8;

res=rt_device_open(can_dev,RT_DEVICE_FLAG_INT_TX|RT_DEVICE_FLAG_INT_RX);
    RT_ASSERT(res==RT_EOK);

thread_can_rx=rt_thread_create("can_rx",can_rx_thread,RT_NULL,1024,25,10);
    if(thread_can_rx!=RT_NULL)
    {
        rt_thread_startup(thread_can_rx);
    }
    else
    {
        rt_kprintf("creat can_rx thread failed!\n");
    }

/*-------------tcp client主循环-------------------*/

uint8_t buf[8] = {0,0,0,0,0,0,0,0};
    rt_uint8_t cnt = 0;

maxfdp1 =   sock+1;
    tim.tv_sec=0;
    tim.tv_usec=100;

while(1)
    {
    sock=-1;
    while(sock==-1)
    {
        sock=socket(AF_INET,SOCK_STREAM,0);
        int mode = 0;  //阻塞模式
		ioctlsocket(sock, FIONBIO, &mode);
    }
    rt_kprintf("socket is created.\r\n");

server_addr.sin_family=AF_INET;
    server_addr.sin_port=htons(port);
    server_addr.sin_addr=*((struct  in_addr*)host->h_addr);
    rt_memset(&(server_addr.sin_zero),0,sizeof(server_addr.sin_zero));

// i=-1;
    // while(i==-1)
    // {
    //     i=connect(sock,(struct sockaddr*)&server_addr,sizeof(struct    sockaddr));
    // }
    // rt_kprintf("tcp is connected.\r\n");
    if (connect(sock, (struct sockaddr*)&server_addr, sizeof(struct sockaddr))==-1)
    {
        rt_kprintf("Connect fail!\n");
        closesocket(sock);
        // rt_free(tcp_recv_data);
        // return;
        continue;
    }
    else
    {
        rt_kprintf("Connect successful\n");
    }

while (1)
    {
        FD_ZERO(&readset);
        FD_SET(sock,&readset);
        i=select(maxfdp1,&readset,0,0,&tim);
        if(i==0)
            continue;
        if(FD_ISSET(sock,&readset))
        {
            bytes_received=recv(sock,tcp_recv_data,BUFSZ-1,MSG_DONTWAIT);
            if (bytes_received<0)
            {
                closesocket(sock);
                rt_kprintf("\nreceived  error-1,  close   the socket.\r\n");
                rt_free(tcp_recv_data);
                break;
            }
            else    if(bytes_received==0)
            {
                closesocket(sock);
                rt_kprintf("\nreceived  error0,  close   the socket.\r\n");
                rt_free(tcp_recv_data);
                break;
            }
            tcp_recv_data[bytes_received]='\0';
            {
                // rt_kprintf("thread1: recv msg from msg queue, the content:%c\n", tcp_recv_data[0]);
                rt_kprintf("tcp_thread: recv data = %s\n", tcp_recv_data);
                can_msg.data[0]=tcp_recv_data[0];
                can_msg.data[1]=tcp_recv_data[1];
                can_msg.data[2]=tcp_recv_data[2];
                can_msg.data[3]=tcp_recv_data[3];
                can_msg.data[4]=tcp_recv_data[4];
                can_msg.data[5]=tcp_recv_data[5];
                can_msg.data[6]=tcp_recv_data[6];
                can_msg.data[7]=tcp_recv_data[7];
                can_size=rt_device_write(can_dev,0,&can_msg,sizeof(can_msg));
            }
        }

/* 延时 50ms */
        rt_thread_mdelay(50);
    }
    }
}

/*--------------------------------------*/

/* 导出到 msh 命令列表中 */
MSH_CMD_EXPORT(can2tcp, can to tcp client sample);
```

这是群里朋友的程序
```c
#include <rtthread.h>
#include "rtdevice.h"

#include <lwip/api.h>
#include <lwip/sockets.h>
#include <lwip/netdb.h>

// #include    <sys/select.h>
// #include    <sys/socket.h>
// #include    <netdb.h>
#include    <string.h>
#include    <finsh.h>

#define CAN_DEV_NAME    "can1"

/* 邮箱控制块 */
static struct rt_mailbox can_mb;
/* 用于放邮件的内存池 */
static char can_mb_pool[256] = {0};

/* 消息队列控制块 */
static struct rt_messagequeue tcp2can_mq;
/* 消息队列中用到的放置消息的内存池 */
static rt_uint8_t tcp2can_msg_pool[2048];

char    can_name[RT_NAME_MAX];
static  rt_device_t can_dev;

int can_msg_send(uint8_t *data,uint16_t len)
{
	rt_uint32_t mailbox_rx_data = 0x20000000;
	uint8_t send_len = 0;
	rt_err_t res;
	while(len > 0 )
	{
		if(len > 8)
		{
			send_len = 8;
		}else
		{
			send_len = len;
		}
		res = rt_mq_send(&tcp2can_mq, data, send_len);//数据存入 tcp2can_mq
		rt_mb_send(&can_mb,mailbox_rx_data);  //通知线程处理
		
		data += send_len;
		len -= send_len;
	}
}

static  rt_err_t can_rx_call(rt_device_t dev, rt_size_t size)
{
	rt_uint32_t mailbox_rx_data = 0x10000000;
	rt_mb_send(&can_mb,mailbox_rx_data); //通知线程处理
    return  RT_EOK;
}

extern int tcp_msg_send(uint8_t *data,uint16_t len);

static void can_thread(void  *parameter)
{
	rt_err_t result;
    /* 初始化一个 mailbox */
    result = rt_mb_init(&can_mb,
                        "can_mb",                      /* 名称是 can_mb */
                        &can_mb_pool[0],                /* 邮箱用到的内存池是 mb_pool */
                        sizeof(can_mb_pool) / 4,        /* 邮箱中的邮件数目，因为一封邮件占 4 字节 */
                        RT_IPC_FLAG_FIFO);          /* 采用 FIFO 方式进行线程等待 */
						
	/* 初始化消息队列 */
    result = rt_mq_init(&tcp2can_mq,
                        "tcp2can_mq",
                        &tcp2can_msg_pool[0],             /* 内存池指向 msg_pool */
                        8,                          /* 每个消息的大小是 8 字节 */
                        sizeof(tcp2can_msg_pool),        /* 内存池的大小是 msg_pool 的大小 */
                        RT_IPC_FLAG_FIFO);       /* 如果有多个线程等待，按照先来先得到的方法分配消息 */					
	
	/*-------------打开can线程----------------*/
    rt_strncpy(can_name,CAN_DEV_NAME,RT_NAME_MAX);
    can_dev =   rt_device_find(can_name);
    if(!can_dev)
    {
        rt_kprintf("find %s  failed!\n",can_name);
        return;
    }

rt_thread_mdelay(3000);

rt_device_open(can_dev, RT_DEVICE_FLAG_INT_TX|RT_DEVICE_FLAG_INT_RX);
    rt_device_set_rx_indicate(can_dev,  can_rx_call);
	
	int rc = 0;
	int size = 0;
    struct rt_can_msg  can_rx_msg = {0};
    struct rt_can_msg  can_tx_msg = {0};
    rt_int8_t i;
	
	rt_uint32_t mailbox_rx_data = 0;
    while(1)
    {
		/* 从邮箱中收取邮件 */
		if (rt_mb_recv(&can_mb, (rt_uint32_t *)&mailbox_rx_data, RT_WAITING_FOREVER) == RT_EOK)
        {
			if(mailbox_rx_data == 0x10000000)//表示CAN接受
			{
				can_rx_msg.hdr = -1; /* hdr 值为 - 1，表示直接从 uselist 链表读取数据 */
				rc = rt_device_read(can_dev,0,&can_rx_msg,sizeof(can_rx_msg));
				if(rc > 0)
				{
					rt_kprintf("ID:%x",can_rx_msg.id);
					for(i=0;i<8;i++)
					{
						rt_kprintf("%2x",can_rx_msg.data[i]);
					}
					rt_kprintf("\n");
					tcp_msg_send(can_rx_msg.data,8);  //发送给tcp
				}
			}
			if(mailbox_rx_data == 0x20000000)//表示CAN发送
			{
				/* 从消息队列中接收消息 */ 
				if (rt_mq_recv(&tcp2can_mq, (void*)can_tx_msg.data, 8, 0) == RT_EOK) //不堵塞接收
				{
					can_tx_msg.id = 0x78;              /* ID 为 0x78 */
					can_tx_msg.ide = RT_CAN_STDID;     /* 标准格式 */
					can_tx_msg.rtr = RT_CAN_DTR;       /* 数据帧 */
					can_tx_msg.len = 8;                /* 数据长度为 8 */
					/* 发送一帧 CAN 数据 */
					size = rt_device_write(can_dev, 0, &can_tx_msg, sizeof(can_tx_msg));
					if (size == 0)
					{
						rt_kprintf("can dev write data failed!\n");
					}
				}
			}
		}
    }
}

#define HOST "192.168.1.143"
#define PORT "5000"
static  int socketfd = -1;

int tcp_msg_send(uint8_t *data,uint16_t len)
{
	if(socketfd > 0)
	{
		send(socketfd, data, len, 0);
	}
}

int tcp_connect1(const char *host, const char *port)
{
	int socket_fd = -1;
	struct addrinfo hints, *addr_list, *cur;
	/* Do name resolution with both IPv6 and IPv4 */
	memset(&hints, 0, sizeof(hints));
	hints.ai_family = AF_INET;
	hints.ai_socktype = SOCK_STREAM;
	hints.ai_protocol = IPPROTO_TCP;
	//hints.ai_protocol = PROTOCOL_TLS;
    rt_err_t ret;

ret = getaddrinfo(host, port, &hints, &addr_list);
	if (ret != 0)
	{
			rt_kprintf("getaddrinfo fail [%d]\n",ret);
			return ret;
	}
	for (cur = addr_list; cur != NULL; cur = cur->ai_next)
	{
		socket_fd = socket(cur->ai_family,SOCK_STREAM, IPPROTO_TCP);
		if (socket_fd < 0) 
		{
			continue;
		}
		int mode = 0;  //阻塞模式
		ioctlsocket(socket_fd, FIONBIO, &mode);
		ret = connect(socket_fd, cur->ai_addr, cur->ai_addrlen);
		rt_kprintf("socket:%d  ret:%d\n",socket_fd,ret);
		if (ret == 0)
		{
			break;//连接成功
		}
		closesocket(socket_fd);
	}
	 freeaddrinfo(addr_list);
	 
	 return socket_fd;
}

static void tcp_rx_thread(void  *parameter)
{
    int maxfdp1;
    fd_set readset;
	struct timeval timeout;
	int rc = 0;
	
	uint8_t *tcp_recv_data = rt_malloc(1024);
	/*-------------tcp client主循环-------------------*/
    while (1)
    {
		socketfd = tcp_connect1(HOST,PORT);  //tcp 
		while(socketfd  > 0)
		{
			FD_ZERO(&readset);
			FD_SET(socketfd,&readset);
			
			maxfdp1 = socketfd+1;
			
			timeout.tv_sec=1;
			timeout.tv_usec=1000;
			
			rc = select(maxfdp1,&readset,0,0,&timeout);
			if(rc==0)
				continue;
			if(FD_ISSET(socketfd,&readset))
			{
                rt_kprintf("开始接受\r\n");
				rc = recv(socketfd,tcp_recv_data,1024,MSG_DONTWAIT);
				if(rc <= 0)
				{
                    rt_kprintf("接受出错\r\n");
					if(!(errno == EINTR || errno == EWOULDBLOCK || errno == EAGAIN))
					{
						rt_kprintf("socket recv err :%d\n",rc);
						closesocket(socketfd);
						socketfd = -1;
						break;
					}else
					{
						rt_kprintf("socket recv timeout :%d\n",rc);
						
					}
				}else
				{
					rt_kprintf("socket recv :%d\n",rc);
					// can_msg_send(tcp_recv_data,rc);
				}
			}
		}
        rt_thread_mdelay(5000); //断线5s后重连
    }
	rt_free(tcp_recv_data);
}

/* 线程 1 入口函数 */
void can2tcp(int  argc,char *argv[])
{
    // rt_thread_t can_tid;
    // can_tid = rt_thread_create("can_task",can_thread,RT_NULL,1024,25,10);
    // if(can_tid!=RT_NULL)
    // {
    //     rt_thread_startup(can_tid);
    // }
    // else
    // {
    //     rt_kprintf("create can_task thread failed!\n");
    // }
	
    rt_thread_t tcp_tid;
    tcp_tid = rt_thread_create("tcp_task",tcp_rx_thread,RT_NULL,1024,25,10);
    if(tcp_tid!=RT_NULL)
    {
        rt_thread_startup(tcp_tid);
    }
    else
    {
        rt_kprintf("create tcp_task thread failed!\n");
    }
}
/*--------------------------------------*/
/* 导出到 msh 命令列表中 */
MSH_CMD_EXPORT(can2tcp, can to tcp client select sample);

```

想请大家分析下tcp重连的方法。