第 4 篇 : Netty客户端互发图片和音/视频

说明

因为图片和音/视频不能确定其具体大小, 故引入MinIO。客户端之间只发送消息, 通过上传/下载来获取额外信息

1. MinIO搭建(参考前面文章), 并启动

2. 登录MinIO创建3个Bucket: image、voice、video

创建桶
3个桶

3. 客户端改造

3.1 修改 pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.3.4.RELEASE</version><relativePath/> <!-- lookup parent from repository --></parent><groupId>com.hahashou.netty</groupId><artifactId>client</artifactId><version>1.0-SNAPSHOT</version><name>client</name><description>Netty Client Project For Spring Boot</description><properties><java.version>1.8</java.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency><dependency><groupId>io.netty</groupId><artifactId>netty-all</artifactId><version>4.1.100.Final</version></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.58</version></dependency><dependency><groupId>com.hahashou</groupId><artifactId>minio</artifactId><version>2023.11.16</version></dependency><!-- 此包在引入minio时必须添加--><dependency><groupId>com.squareup.okhttp3</groupId><artifactId>okhttp</artifactId><version>4.9.3</version></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId><configuration><excludes><exclude><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId></exclude></excludes></configuration></plugin></plugins></build></project>

3.2 修改 application.yml, 限制单文件大小为128M以内, 总大小为256M以内

server:port: 32001logging:level:com.hahashou.netty: infospring:servlet:multipart:max-file-size: 128MBmax-request-size: 256MBuserCode: Aaminio:endpoint: http://127.0.0.1:9000accessKey: rootsecretKey: root123456

3.3 config包下新增 MinIOConfig类

package com.hahashou.netty.client.config;import io.minio.MinioClient;
import io.minio.http.HttpUtils;
import okhttp3.OkHttpClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*** @description: MinIO配置* @author: 哼唧兽* @date: 9999/9/21**/
@Configuration
public class MinIOConfig {@Value("${minio.endpoint}")private String endpoint;@Value("${minio.accessKey}")private String accessKey;@Value("${minio.secretKey}")private String secretKey;@Beanpublic MinioClient minioClient() {OkHttpClient okHttpClient = HttpUtils.newDefaultHttpClient(60 * 1000,50 * 1000, 30 * 1000);MinioClient minioClient = MinioClient.builder().endpoint(endpoint).credentials(accessKey, secretKey).httpClient(okHttpClient).build();return minioClient;}
}

3.4 新增utils包, 放入 MinioUtils类

package com.hahashou.netty.client.utils;import io.minio.*;
import io.minio.errors.MinioException;
import io.minio.http.Method;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import org.springframework.web.multipart.MultipartFile;import javax.annotation.Resource;
import java.io.File;
import java.io.IOException;
import java.security.InvalidKeyException;
import java.security.NoSuchAlgorithmException;
import java.util.concurrent.TimeUnit;/*** @description:* @author: 哼唧兽* @date: 9999/9/21**/
@Component
@Slf4j
public class MinioUtils {@Resourceprivate MinioClient minioClient;/*** 文件上传* @param bucketName* @param fileName* @param multipartFile* @return*/public void upload(String bucketName, String multiLevelFolders, String fileName, MultipartFile multipartFile) {try {boolean found = minioClient.bucketExists(BucketExistsArgs.builder().bucket(bucketName).build());if (!found) {minioClient.makeBucket(MakeBucketArgs.builder().bucket(bucketName).build());}String suffix = fileName.substring(fileName.lastIndexOf("."));File file = File.createTempFile(fileName, suffix);multipartFile.transferTo(file);UploadObjectArgs objectArgs = UploadObjectArgs.builder().bucket(bucketName).object(StringUtils.hasText(multiLevelFolders) ? multiLevelFolders + fileName : fileName).filename(file.getAbsolutePath()).contentType(multipartFile.getContentType()).build();minioClient.uploadObject(objectArgs);} catch (MinioException | InvalidKeyException | IOException | NoSuchAlgorithmException exception) {log.error("MinIO文件上传异常 : {}", exception.getMessage());}}/*** 预览地址* @param fileName* @param bucketName* @return*/public String preview(String fileName, String bucketName){// 查看文件地址GetPresignedObjectUrlArgs build = GetPresignedObjectUrlArgs.builder().bucket(bucketName).expiry(30, TimeUnit.MINUTES).object(fileName).method(Method.GET).build();try {return minioClient.getPresignedObjectUrl(build);} catch (Exception exception) {log.error("获取MinIO预览地址异常 : {}", exception.getMessage());}return null;}
}

3.5 修改 Message类

package com.hahashou.netty.client.config;import com.alibaba.fastjson.JSON;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.util.CharsetUtil;
import lombok.Data;
import lombok.Getter;import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;/*** @description:* @author: 哼唧兽* @date: 9999/9/21**/
@Data
public class Message {/** 发送者用户code */private String userCode;/** 接收者用户code */private String friendUserCode;/** 连接时专用 */private String channelId;/** 消息类型 */private Integer type;public enum TypeEnum {TEXT(0, "文字", "", new ArrayList<>()),IMAGE(1, "图片", "image", Arrays.asList("bmp", "gif", "jpeg", "jpg", "png")),VOICE(2, "语音", "voice", Arrays.asList("mp3", "amr", "flac", "wma", "aac")),VIDEO(3, "视频", "video", Arrays.asList("mp4", "avi", "rmvb", "flv", "3gp", "ts", "mkv")),;@Getterprivate Integer key;@Getterprivate String describe;@Getterprivate String bucketName;@Getterprivate List<String> formatList;TypeEnum(int key, String describe, String bucketName, List<String> formatList) {this.key = key;this.describe = describe;this.bucketName = bucketName;this.formatList = formatList;}public static TypeEnum select(String format) {TypeEnum result = null;for (TypeEnum typeEnum : TypeEnum.values()) {if (typeEnum.getFormatList().contains(format)) {result = typeEnum;break;}}return result;}}/** 文字或文件的全路径名称 */private String text;public static ByteBuf transfer(Message message) {return Unpooled.copiedBuffer(JSON.toJSONString(message), CharsetUtil.UTF_8);}/*** 生成指定长度的随机字符串* @param length* @return*/public static String randomString (int length) {if (length > 64) {length = 64;}List<String> list = new ArrayList<>();for (int i = 0; i < 10; i++) {list.add(i + "");}for (char i = 'A'; i <= 'Z'; i++) {list.add(String.valueOf(i));}for (char i = 'a'; i <= 'z'; i++) {list.add(String.valueOf(i));}list.add("α");list.add("ω");Collections.shuffle(list);String string = list.toString();return string.replace("[", "").replace("]", "").replace(", ", "").substring(0, length);}
}

3.6 新增service包, 放入 ClientService接口

package com.hahashou.netty.client.service;import com.hahashou.netty.client.config.Message;import javax.servlet.http.HttpServletRequest;/*** @description:* @author: 哼唧兽* @date: 9999/9/21**/
public interface ClientService {/*** 上传* @param userCode* @param httpServletRequest* @return*/Message upload(String userCode, HttpServletRequest httpServletRequest);/*** 下载链接* @param fileName* @return*/String link(String fileName);
}

3.7 service包下新建impl包, 放入 ClientServiceImpl类

package com.hahashou.netty.client.service.imol;import com.hahashou.netty.client.config.Message;
import com.hahashou.netty.client.service.ClientService;
import com.hahashou.netty.client.utils.MinioUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.web.multipart.MultipartFile;
import org.springframework.web.multipart.MultipartHttpServletRequest;import javax.annotation.Resource;
import javax.servlet.http.HttpServletRequest;
import java.time.LocalDate;
import java.time.LocalTime;
import java.util.*;/*** @description:* @author: 哼唧兽* @date: 9999/9/21**/
@Service
@Slf4j
public class ClientServiceImpl implements ClientService {@Resourceprivate MinioUtils minioUtils;@Overridepublic Message upload(String userCode, HttpServletRequest httpServletRequest) {Message result = new Message();MultipartHttpServletRequest multipartHttpServletRequest = (MultipartHttpServletRequest) httpServletRequest;List<MultipartFile> multipartFileList = multipartHttpServletRequest.getFiles("file");if (!CollectionUtils.isEmpty(multipartFileList)) {MultipartFile multipartFile = multipartFileList.get(0);String originalFilename = multipartFile.getOriginalFilename();String suffix = originalFilename.substring(originalFilename.lastIndexOf(".") + 1);Message.TypeEnum typeEnum = Message.TypeEnum.select(suffix);if (typeEnum != null) {String fileName = generateFileName(suffix);String multiLevelFolders = userCode + "/"+ LocalDate.now().toString().replace("-", "") + "/";minioUtils.upload(typeEnum.getBucketName(), multiLevelFolders, fileName, multipartFile);result.setType(typeEnum.getKey());result.setText(multiLevelFolders + fileName);}}return result;}public static String generateFileName(String suffix) {LocalTime now = LocalTime.now();return now.toString().replace(":", "").replace(".", "-") + Message.randomString(4) + "." + suffix;}@Overridepublic String link(String fileName) {String suffix = fileName.substring(fileName.lastIndexOf(".") + 1);Message.TypeEnum typeEnum = Message.TypeEnum.select(suffix);if (typeEnum != null) {return minioUtils.preview(fileName, typeEnum.getBucketName());}return null;}
}

3.8 修改 ClientController类

package com.hahashou.netty.client.controller;import com.alibaba.fastjson.JSON;
import com.hahashou.netty.client.config.Message;
import com.hahashou.netty.client.config.NettyClientHandler;
import com.hahashou.netty.client.service.ClientService;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.*;import javax.annotation.Resource;
import javax.servlet.http.HttpServletRequest;/*** @description:* @author: 哼唧兽* @date: 9999/9/21**/
@RestController
@RequestMapping("/client")
@Slf4j
public class ClientController {@Resourceprivate ClientService clientService;@PostMapping("/send")public String send(@RequestBody Message dto) {Channel channel = NettyClientHandler.CHANNEL;if (channel == null) {return "服务端已下线";}channel.writeAndFlush(Message.transfer(dto));return "success";}@GetMapping("/upload/{userCode}")public String upload(@PathVariable String userCode, final HttpServletRequest httpServletRequest) {if (StringUtils.isEmpty(userCode)) {return "userCode is null";}Message upload = clientService.upload(userCode, httpServletRequest);return JSON.toJSONString(upload);}@GetMapping("/link")public String link(@RequestParam String fileName) {// 如果Bucket包含多级目录, fileName为Bucket下文件的全路径名if (StringUtils.isEmpty(fileName)) {return "fileName is null";}return clientService.link(fileName);}
}

4. 服务端改造

4.1 复制客户端的 Message类

4.2 修改 NettyServerHandler类

package com.hahashou.netty.server.config;import com.alibaba.fastjson.JSON;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;/*** @description:* @author: 哼唧兽* @date: 9999/9/21**/
@Component
@ChannelHandler.Sharable
@Slf4j
public class NettyServerHandler extends ChannelInboundHandlerAdapter {/** key: 用户code; value: channelId */public static Map<String, String> USER_CHANNEL = new ConcurrentHashMap<>(32);/** key: channelId; value: Channel */public static Map<String, Channel> CHANNEL = new ConcurrentHashMap<>(32);/** 用户离线消息 */public static Map<String, List<Message>> USER_MESSAGE = new ConcurrentHashMap<>(32);@Overridepublic void channelActive(ChannelHandlerContext ctx) {Channel channel = ctx.channel();String channelId = channel.id().asLongText();log.info("有客户端连接, channelId : {}", channelId);CHANNEL.put(channelId, channel);Message message = new Message();message.setChannelId(channelId);channel.writeAndFlush(Message.transfer(message));}@Overridepublic void channelInactive(ChannelHandlerContext ctx) {log.info("有客户端断开连接, channelId : {}", ctx.channel().id().asLongText());CHANNEL.remove(ctx.channel().id().asLongText());}@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) {if (msg != null) {Message message = JSON.parseObject(msg.toString(), Message.class);String userCode = message.getUserCode(),channelId = message.getChannelId();if (StringUtils.hasText(userCode) && StringUtils.hasText(channelId)) {connect(userCode, channelId);} else if (StringUtils.hasText(message.getText())) {if (StringUtils.hasText(message.getFriendUserCode())) {sendOtherClient(message);} else {sendAdmin(ctx.channel(), message);}}}}/*** 建立连接* @param userCode* @param channelId*/private void connect(String userCode, String channelId) {log.info("客户端 {} 连接", userCode);USER_CHANNEL.put(userCode, channelId);}/*** 发送给其他客户端* @param message*/private void sendOtherClient(Message message) {String friendUserCode = message.getFriendUserCode();String queryChannelId = USER_CHANNEL.get(friendUserCode);if (StringUtils.hasText(queryChannelId)) {Channel channel = CHANNEL.get(queryChannelId);if (channel == null) {offlineMessage(friendUserCode, message);return;}channel.writeAndFlush(Message.transfer(message));} else {offlineMessage(friendUserCode, message);}}/*** 离线消息存储* @param friendUserCode* @param message*/public void offlineMessage(String friendUserCode, Message message) {List<Message> messageList = USER_MESSAGE.get(friendUserCode);if (CollectionUtils.isEmpty(messageList)) {messageList = new ArrayList<>();}messageList.add(message);USER_MESSAGE.put(friendUserCode, messageList);}/*** 发送给服务端* @param channel* @param message*/private void sendAdmin(Channel channel, Message message) {message.setUserCode("ADMIN");message.setText(LocalDateTime.now().toString());channel.writeAndFlush(Message.transfer(message));}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {log.info("有客户端发生异常, channelId : {}", ctx.channel().id().asLongText());}
}