Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[ROCKETMQ-13] Wrong log level for AcceptSocketService termination. #12

Closed
wants to merge 1 commit into from
Closed

[ROCKETMQ-13] Wrong log level for AcceptSocketService termination. #12

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
26 changes: 17 additions & 9 deletions rocketmq-store/src/main/java/com/alibaba/rocketmq/store/ha/HAService.java
View file
Open in desktop
Expand Up @@ -46,7 +46,7 @@ public class HAService {

private final AtomicInteger connectionCount = new AtomicInteger(0);

private final List<HAConnection> connectionList = new LinkedList<HAConnection>();
private final List<HAConnection> connectionList = new LinkedList<>();

private final AcceptSocketService acceptSocketService;

Expand Down Expand Up @@ -170,17 +170,22 @@ public AtomicLong getPush2SlaveMaxOffset() {
return push2SlaveMaxOffset;
}

/**
* Listens to slave connections to create {@link HAConnection}.
*/
class AcceptSocketService extends ServiceThread {
private ServerSocketChannel serverSocketChannel;
private Selector selector;
private final SocketAddress socketAddressListen;


public AcceptSocketService(final int port) {
this.socketAddressListen = new InetSocketAddress(port);
}


/**
* Starts listening to slave connections.
* @throws Exception If fails.
*/
public void beginAccept() throws Exception {
this.serverSocketChannel = ServerSocketChannel.open();
this.selector = RemotingUtil.openSelector();
Expand All @@ -190,6 +195,7 @@ public void beginAccept() throws Exception {
this.serverSocketChannel.register(this.selector, SelectionKey.OP_ACCEPT);
}

/** {@inheritDoc} */
@Override
public void shutdown(final boolean interrupt) {
super.shutdown(interrupt);
Expand All @@ -202,6 +208,7 @@ public void shutdown(final boolean interrupt) {
}
}

/** {@inheritDoc} */
@Override
public void run() {
log.info(this.getServiceName() + " service started");
Expand All @@ -210,10 +217,12 @@ public void run() {
try {
this.selector.select(1000);
Set<SelectionKey> selected = this.selector.selectedKeys();

if (selected != null) {
for (SelectionKey k : selected) {
if ((k.readyOps() & SelectionKey.OP_ACCEPT) != 0) {
SocketChannel sc = ((ServerSocketChannel) k.channel()).accept();

if (sc != null) {
HAService.log.info("HAService receive new connection, "
+ sc.socket().getRemoteSocketAddress());
Expand All @@ -234,16 +243,15 @@ public void run() {

selected.clear();
}

} catch (Exception e) {
log.error(this.getServiceName() + " service has exception.", e);
}
}

log.error(this.getServiceName() + " service end");
log.info(this.getServiceName() + " service end");
}


/** {@inheritDoc} */
@Override
public String getServiceName() {
return AcceptSocketService.class.getSimpleName();
Expand All @@ -256,8 +264,8 @@ public String getServiceName() {
class GroupTransferService extends ServiceThread {

private final WaitNotifyObject notifyTransferObject = new WaitNotifyObject();
private volatile List<GroupCommitRequest> requestsWrite = new ArrayList<GroupCommitRequest>();
private volatile List<GroupCommitRequest> requestsRead = new ArrayList<GroupCommitRequest>();
private volatile List<GroupCommitRequest> requestsWrite = new ArrayList<>();
private volatile List<GroupCommitRequest> requestsRead = new ArrayList<>();


public void putRequest(final GroupCommitRequest request) {
Expand Down Expand Up @@ -333,7 +341,7 @@ public String getServiceName() {

class HAClient extends ServiceThread {
private static final int READ_MAX_BUFFER_SIZE = 1024 * 1024 * 4;
private final AtomicReference<String> masterAddress = new AtomicReference<String>();
private final AtomicReference<String> masterAddress = new AtomicReference<>();
private final ByteBuffer reportOffset = ByteBuffer.allocate(8);
private SocketChannel socketChannel;
private Selector selector;
Expand Down