Добавлена многопоточность
This commit is contained in:
parent
bac5ed4dc1
commit
0f170b8691
GPG Key ID:
6D45AA2C8FD6A37E
@ -18,34 +18,28 @@ import org.slf4j.LoggerFactory;
|
||||
import java.io.ByteArrayInputStream;
|
||||
import java.io.ByteArrayOutputStream;
|
||||
import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
import java.io.ObjectInputStream;
|
||||
import java.io.ObjectOutputStream;
|
||||
import java.io.Serializable;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.net.SocketAddress;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.nio.channels.SelectionKey;
|
||||
import java.nio.channels.Selector;
|
||||
import java.nio.channels.ServerSocketChannel;
|
||||
import java.nio.channels.SocketChannel;
|
||||
import java.net.ServerSocket;
|
||||
import java.net.Socket;
|
||||
import java.util.ArrayList;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ExecutionException;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
|
||||
public class Server {
|
||||
private static final Logger log = LoggerFactory.getLogger(Server.class);
|
||||
private final Selector selector;
|
||||
private final ServerSocketChannel serverSocketChannel;
|
||||
private final Map<SocketAddress, Message> usersMessages = new HashMap<>();
|
||||
private final ServerSocket serverSocket;
|
||||
private final IStorage storage;
|
||||
private boolean isRunning = false;
|
||||
private boolean isRunning;
|
||||
private final ExecutorService requestReadPool = Executors.newCachedThreadPool();
|
||||
private final ExecutorService requestProcessPool = Executors.newCachedThreadPool();
|
||||
private final ExecutorService responseSendPool = Executors.newFixedThreadPool(12);
|
||||
|
||||
public Server(int port, IStorage collection) throws IOException {
|
||||
selector = Selector.open();
|
||||
serverSocketChannel = ServerSocketChannel.open();
|
||||
serverSocketChannel.bind(new InetSocketAddress(port));
|
||||
serverSocketChannel.configureBlocking(false);
|
||||
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
|
||||
serverSocket = new ServerSocket(port);
|
||||
isRunning = true;
|
||||
|
||||
if (collection == null) collection = new ProductCollection(new ArrayList<>());
|
||||
@ -54,80 +48,70 @@ public class Server {
|
||||
log.info("The server has been assigned to port {}", port);
|
||||
}
|
||||
|
||||
public static ByteBuffer serialize(Serializable obj) throws IOException {
|
||||
public static byte[] serialize(Serializable obj) throws IOException {
|
||||
try (var bOut = new ByteArrayOutputStream();
|
||||
var oOut = new ObjectOutputStream(bOut)) {
|
||||
oOut.writeObject(obj);
|
||||
oOut.flush();
|
||||
return ByteBuffer.wrap(bOut.toByteArray());
|
||||
return bOut.toByteArray();
|
||||
}
|
||||
}
|
||||
|
||||
public void run() throws IOException {
|
||||
public void run() {
|
||||
log.info("The server is running");
|
||||
|
||||
while (isRunning) {
|
||||
selector.select(this::handleKey);
|
||||
try {
|
||||
var clientSocket = serverSocket.accept();
|
||||
log.info("Received connection from {}", clientSocket.getRemoteSocketAddress());
|
||||
requestReadPool.execute(() -> {
|
||||
try {
|
||||
handleClient(clientSocket);
|
||||
} catch (IOException | ClassNotFoundException e) {
|
||||
log.error(e.getMessage(), e);
|
||||
}
|
||||
});
|
||||
} catch (IOException e) {
|
||||
log.error(e.getMessage(), e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void handleKey(SelectionKey key) {
|
||||
private void handleClient(Socket clientSocket) throws IOException, ClassNotFoundException {
|
||||
var inputStream = clientSocket.getInputStream();
|
||||
|
||||
var buffer = new byte[8192];
|
||||
inputStream.read(buffer);
|
||||
|
||||
var oIn = new ObjectInputStream(new ByteArrayInputStream(buffer));
|
||||
var msg = (Message) oIn.readObject();
|
||||
log.info("Received message {} from {}", msg, clientSocket.getRemoteSocketAddress());
|
||||
|
||||
responseSendPool.execute(() -> handleWrite(clientSocket, msg, inputStream));
|
||||
}
|
||||
|
||||
private void handleWrite(Socket clientSocket, Message msg, InputStream inputStream) {
|
||||
try {
|
||||
if (key.isAcceptable()) handleAccept(key);
|
||||
if (key.isReadable()) handleRead(key);
|
||||
if (key.isWritable()) handleWrite(key);
|
||||
} catch (IOException | ClassNotFoundException e) {
|
||||
var response = requestProcessPool.submit(() -> createResponse(msg));
|
||||
var responseBuffer = response.get();
|
||||
if (responseBuffer != null) {
|
||||
try {
|
||||
var outputStream = clientSocket.getOutputStream();
|
||||
outputStream.write(responseBuffer);
|
||||
outputStream.flush();
|
||||
log.info("Sent message to {}", clientSocket.getRemoteSocketAddress());
|
||||
outputStream.close();
|
||||
} catch (IOException e) {
|
||||
log.error(e.getMessage(), e);
|
||||
}
|
||||
}
|
||||
inputStream.close();
|
||||
} catch (IOException | InterruptedException | ExecutionException e) {
|
||||
log.error(e.getMessage(), e);
|
||||
key.cancel();
|
||||
}
|
||||
}
|
||||
|
||||
private void handleAccept(SelectionKey key) throws IOException {
|
||||
var server = (ServerSocketChannel) key.channel();
|
||||
var client = server.accept();
|
||||
if (client != null) {
|
||||
client.configureBlocking(false);
|
||||
client.register(selector, SelectionKey.OP_READ);
|
||||
log.info("Received connection from {}", client.getRemoteAddress());
|
||||
}
|
||||
}
|
||||
|
||||
private void handleRead(SelectionKey key) throws IOException, ClassNotFoundException {
|
||||
var client = (SocketChannel) key.channel();
|
||||
var buffer = ByteBuffer.allocate(8192);
|
||||
int bytesRead = client.read(buffer);
|
||||
if (bytesRead == -1) {
|
||||
client.close();
|
||||
} else {
|
||||
buffer.flip();
|
||||
var oIn = new ObjectInputStream(new ByteArrayInputStream(buffer.array()));
|
||||
var msg = (Message) oIn.readObject();
|
||||
log.info("Received message {} from {}", msg, client.getRemoteAddress());
|
||||
usersMessages.put(client.getRemoteAddress(), msg);
|
||||
client.register(selector, SelectionKey.OP_WRITE);
|
||||
oIn.close();
|
||||
}
|
||||
}
|
||||
|
||||
private void handleWrite(SelectionKey key) throws IOException {
|
||||
var client = (SocketChannel) key.channel();
|
||||
var message = usersMessages.get(client.getRemoteAddress());
|
||||
if (message == null) {
|
||||
log.error("Message for {} not found", client.getRemoteAddress());
|
||||
client.close();
|
||||
return;
|
||||
}
|
||||
|
||||
ByteBuffer responseBuffer = createResponse(message);
|
||||
if (responseBuffer != null) {
|
||||
client.write(responseBuffer);
|
||||
log.info("Sent message to {}", client.getRemoteAddress());
|
||||
}
|
||||
|
||||
client.close();
|
||||
}
|
||||
|
||||
private ByteBuffer createResponse(Message message) throws IOException {
|
||||
private byte[] createResponse(Message message) throws IOException {
|
||||
var user = (User) message.getUser();
|
||||
if (user != null) user.setPassword(SHA256.hash(user.getPassword()));
|
||||
return switch (message.getType()) {
|
||||
@ -141,7 +125,7 @@ public class Server {
|
||||
};
|
||||
}
|
||||
|
||||
private ByteBuffer handleGetCommand(String inputCommand, User user) throws IOException {
|
||||
private byte[] handleGetCommand(String inputCommand, User user) throws IOException {
|
||||
var command = CommandManager.getCommandByInput(inputCommand);
|
||||
if (command.isEmpty()) {
|
||||
log.error("Command {} not found", inputCommand);
|
||||
@ -155,7 +139,7 @@ public class Server {
|
||||
}
|
||||
}
|
||||
|
||||
private ByteBuffer handlePostCommand(MessageBody request, User user) throws IOException {
|
||||
private byte[] handlePostCommand(MessageBody request, User user) throws IOException {
|
||||
var inputCommand = request.getCommand();
|
||||
var body = request.getBody();
|
||||
|
||||
@ -175,7 +159,9 @@ public class Server {
|
||||
|
||||
public void stop() throws IOException {
|
||||
isRunning = false;
|
||||
serverSocketChannel.close();
|
||||
selector.close();
|
||||
serverSocket.close();
|
||||
requestReadPool.shutdown();
|
||||
requestProcessPool.shutdown();
|
||||
responseSendPool.shutdown();
|
||||
}
|
||||
}
|
||||
|
||||
@ -20,14 +20,13 @@ import java.util.Objects;
|
||||
import java.util.Optional;
|
||||
import java.util.Set;
|
||||
import java.util.TreeSet;
|
||||
import java.util.concurrent.locks.ReentrantLock;
|
||||
import java.util.concurrent.locks.Lock;
|
||||
|
||||
/**
|
||||
* Represents a collection of products.
|
||||
*/
|
||||
public class ProductCollection implements IStorage {
|
||||
private static final Logger log = LoggerFactory.getLogger(ProductCollection.class);
|
||||
private TreeSet<Product> productList;
|
||||
private Long idIncrementor;
|
||||
private final Lock lock = new ReentrantLock();
|
||||
|
||||
private List<Product> fetchProductFromDb(Connection connection) {
|
||||
try {
|
||||
@ -45,7 +44,6 @@ public class ProductCollection implements IStorage {
|
||||
return Math.toIntExact((long) (o1Length - o2Length));
|
||||
});
|
||||
productList.addAll(list);
|
||||
idIncrementor = productList.isEmpty() ? 1L : productList.stream().map(Product::getId).max(Long::compare).orElse(1L);
|
||||
}
|
||||
|
||||
public ProductCollection(List<Product> list) {
|
||||
@ -61,84 +59,154 @@ public class ProductCollection implements IStorage {
|
||||
}
|
||||
|
||||
public Product getProductById(Long id) {
|
||||
Optional<Product> product = productList.stream().filter(i -> Objects.equals(i.getId(), id)).findFirst();
|
||||
return product.orElse(null);
|
||||
lock.lock();
|
||||
try {
|
||||
Optional<Product> product = productList.stream().filter(i -> Objects.equals(i.getId(), id)).findFirst();
|
||||
return product.orElse(null);
|
||||
} finally {
|
||||
lock.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
public Product addProduct(ProductDTO productDTO, User user) throws SQLException {
|
||||
var product = DbWrapper.addProduct(productDTO, DbController.getConnection(), user);
|
||||
productList.add(product);
|
||||
return product;
|
||||
lock.lock();
|
||||
try {
|
||||
var product = DbWrapper.addProduct(productDTO, DbController.getConnection(), user);
|
||||
productList.add(product);
|
||||
return product;
|
||||
} finally {
|
||||
lock.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
public Product updateProduct(Long id, ProductDTO productDTO, User user) throws SQLException {
|
||||
var product = DbWrapper.updateProduct(id, productDTO, DbController.getConnection(), user);
|
||||
fetchProductFromDb(DbController.getConnection());
|
||||
return productDTO.buildProduct(product.getId(), product.getCreationDate());
|
||||
lock.lock();
|
||||
try {
|
||||
var product = DbWrapper.updateProduct(id, productDTO, DbController.getConnection(), user);
|
||||
fetchProductFromDb(DbController.getConnection());
|
||||
return productDTO.buildProduct(product.getId(), product.getCreationDate());
|
||||
} finally {
|
||||
lock.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
public Product removeProduct(Long id, User user) throws SQLException {
|
||||
var product = getProductById(id);
|
||||
if (DbWrapper.removeProductById(id, DbController.getConnection(), user)) {
|
||||
productList.remove(product);
|
||||
return product;
|
||||
lock.lock();
|
||||
try {
|
||||
var product = getProductById(id);
|
||||
if (DbWrapper.removeProductById(id, DbController.getConnection(), user)) {
|
||||
productList.remove(product);
|
||||
return product;
|
||||
}
|
||||
return null;
|
||||
} finally {
|
||||
lock.unlock();
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
public String getInfo() {
|
||||
var optionalInitDate = productList.stream().map(Product::getCreationDate).sorted().findFirst();
|
||||
ZonedDateTime initDate = optionalInitDate.orElse(ZonedDateTime.ofInstant(Instant.EPOCH, ZoneId.systemDefault()));
|
||||
return String.format("Структура: TreeSet%nДата инициализации: %s%nКоличество элементов: %s", initDate, productList.size());
|
||||
lock.lock();
|
||||
try {
|
||||
var optionalInitDate = productList.stream().map(Product::getCreationDate).sorted().findFirst();
|
||||
ZonedDateTime initDate = optionalInitDate.orElse(ZonedDateTime.ofInstant(Instant.EPOCH, ZoneId.systemDefault()));
|
||||
return String.format("Структура: TreeSet%nДата инициализации: %s%nКоличество элементов: %s", initDate, productList.size());
|
||||
} finally {
|
||||
lock.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
public void clear(User user) throws SQLException {
|
||||
DbWrapper.clearProducts(user, DbController.getConnection());
|
||||
fetchProductFromDb(DbController.getConnection());
|
||||
lock.lock();
|
||||
try {
|
||||
DbWrapper.clearProducts(user, DbController.getConnection());
|
||||
fetchProductFromDb(DbController.getConnection());
|
||||
} finally {
|
||||
lock.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
public Long getMaxPrice() {
|
||||
return productList.stream().max(Comparator.comparingLong(Product::getPrice)).map(Product::getPrice).orElse(null);
|
||||
lock.lock();
|
||||
try {
|
||||
return productList.stream().max(Comparator.comparingLong(Product::getPrice)).map(Product::getPrice).orElse(null);
|
||||
} finally {
|
||||
lock.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
public Long getMinPrice() {
|
||||
return productList.stream().min(Comparator.comparingLong(Product::getPrice)).map(Product::getPrice).orElse(null);
|
||||
lock.lock();
|
||||
try {
|
||||
return productList.stream().min(Comparator.comparingLong(Product::getPrice)).map(Product::getPrice).orElse(null);
|
||||
} finally {
|
||||
lock.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
public boolean isProductIdExists(Long id) {
|
||||
return getProductById(id) != null;
|
||||
lock.lock();
|
||||
try {
|
||||
return getProductById(id) != null;
|
||||
} finally {
|
||||
lock.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
public Integer removeLover(Long id, User user) throws SQLException {
|
||||
var size = productList.size();
|
||||
DbWrapper.removeLower(id, DbController.getConnection(), user);
|
||||
fetchProductFromDb(DbController.getConnection());
|
||||
return size - productList.size();
|
||||
lock.lock();
|
||||
try {
|
||||
var size = productList.size();
|
||||
DbWrapper.removeLower(id, DbController.getConnection(), user);
|
||||
fetchProductFromDb(DbController.getConnection());
|
||||
return size - productList.size();
|
||||
} finally {
|
||||
lock.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
public String filterContainsName(String name) {
|
||||
return String.join("\n", productList.stream()
|
||||
.filter(product -> product.getName().toLowerCase().contains(name.toLowerCase()))
|
||||
.map(Product::toString)
|
||||
.toList());
|
||||
lock.lock();
|
||||
try {
|
||||
return String.join("\n", productList.stream()
|
||||
.filter(product -> product.getName().toLowerCase().contains(name.toLowerCase()))
|
||||
.map(Product::toString)
|
||||
.toList());
|
||||
} finally {
|
||||
lock.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
public String getUniqueManufactureCost() {
|
||||
return String.join(", ", Set.copyOf(productList.stream()
|
||||
.map(Product::getManufactureCost)
|
||||
.toList())
|
||||
.stream()
|
||||
.map(Object::toString)
|
||||
.toList());
|
||||
lock.lock();
|
||||
try {
|
||||
return String.join(", ", Set.copyOf(productList.stream()
|
||||
.map(Product::getManufactureCost)
|
||||
.toList())
|
||||
.stream()
|
||||
.map(Object::toString)
|
||||
.toList());
|
||||
} finally {
|
||||
lock.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
public List<Product> toList() {
|
||||
return productList.stream().toList();
|
||||
lock.lock();
|
||||
try {
|
||||
return productList.stream().toList();
|
||||
} finally {
|
||||
lock.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return String.join("\n", productList.stream().map(Product::toString).toList());
|
||||
lock.lock();
|
||||
try {
|
||||
return String.join("\n", productList.stream().map(Product::toString).toList());
|
||||
} finally {
|
||||
lock.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
@ -146,11 +214,21 @@ public class ProductCollection implements IStorage {
|
||||
if (this == o) return true;
|
||||
if (o == null || getClass() != o.getClass()) return false;
|
||||
ProductCollection that = (ProductCollection) o;
|
||||
return Objects.equals(productList, that.productList) && Objects.equals(idIncrementor, that.idIncrementor);
|
||||
lock.lock();
|
||||
try {
|
||||
return Objects.equals(productList, that.productList);
|
||||
} finally {
|
||||
lock.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public int hashCode() {
|
||||
return Objects.hash(productList, idIncrementor);
|
||||
lock.lock();
|
||||
try {
|
||||
return Objects.hashCode(productList);
|
||||
} finally {
|
||||
lock.unlock();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Loading…
Reference in New Issue
Block a user