更新用户信息时，先写入数据库再存入缓存中，防止异步操作写入数据库失败造成数据不一致

broker/src/main/java/com/xiaoleilu/loServer/action/admin/CreateRobotAction.java

@@ -55,7 +55,15 @@ public class CreateRobotAction extends AdminAction {
                 WFCMessage.User newUser = inputCreateRobot.toUser();
 
 
-                messagesStore.addUserInfo(newUser, inputCreateRobot.getPassword());
+                try {
+                    messagesStore.addUserInfo(newUser, inputCreateRobot.getPassword());
+                } catch (Exception e) {
+                    e.printStackTrace();
+                    response.setStatus(HttpResponseStatus.OK);
+                    RestResult result = RestResult.resultOf(ErrorCode.ERROR_CODE_SERVER_ERROR, e.getMessage());
+                    response.setContent(new Gson().toJson(result));
+                    return true;
+                }
 
                 if (StringUtil.isNullOrEmpty(inputCreateRobot.getOwner())) {
                     inputCreateRobot.setOwner(inputCreateRobot.getUserId());

broker/src/main/java/com/xiaoleilu/loServer/action/admin/CreateUserAction.java

@@ -75,7 +75,16 @@ public class CreateUserAction extends AdminAction {
 
                 newUserBuilder.setUpdateDt(System.currentTimeMillis());
 
-                messagesStore.addUserInfo(newUserBuilder.build(), inputCreateUser.getPassword());
+
+                try {
+                    messagesStore.addUserInfo(newUserBuilder.build(), inputCreateUser.getPassword());
+                } catch (Exception e) {
+                    e.printStackTrace();
+                    response.setStatus(HttpResponseStatus.OK);
+                    RestResult result = RestResult.resultOf(ErrorCode.ERROR_CODE_SERVER_ERROR, e.getMessage());
+                    response.setContent(new Gson().toJson(result));
+                    return true;
+                }
 
                 response.setStatus(HttpResponseStatus.OK);
                 RestResult result = RestResult.ok(new OutputCreateUser(inputCreateUser.getUserId(), inputCreateUser.getName()));

broker/src/main/java/io/moquette/imhandler/ModifyMyInfoHandler.java

@@ -18,6 +18,11 @@ import win.liyufan.im.IMTopic;
 public class ModifyMyInfoHandler extends IMHandler<WFCMessage.ModifyMyInfoRequest> {
     @Override
     public ErrorCode action(ByteBuf ackPayload, String clientID, String fromUser, boolean isAdmin, WFCMessage.ModifyMyInfoRequest request, Qos1PublishHandler.IMCallback callback) {
+        try {
             return m_messagesStore.modifyUserInfo(fromUser, request);
+        } catch (Exception e) {
+            e.printStackTrace();
+            return ErrorCode.ERROR_CODE_SERVER_ERROR;
+        }
     }
 }

broker/src/main/java/io/moquette/persistence/DatabaseStore.java

@@ -2128,7 +2128,7 @@ public class DatabaseStore {
         });
     }
 
-    void updateUser(final WFCMessage.User user) {
+    void updateUser(final WFCMessage.User user) throws Exception {
         LOG.info("Database update user info {} {}", user.getUid(), user.getUpdateDt());
             Connection connection = null;
             PreparedStatement statement = null;
@@ -2202,6 +2202,7 @@ public class DatabaseStore {
                 // TODO Auto-generated catch block
                 e.printStackTrace();
                 Utility.printExecption(LOG, e);
+                throw new Exception(e.getMessage());
             } finally {
                 DBUtil.closeDB(connection, statement);
             }

broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java

@@ -1417,7 +1417,7 @@ public class MemoryMessagesStore implements IMessagesStore {
     }
 
     @Override
-    public ErrorCode modifyUserInfo(String userId, WFCMessage.ModifyMyInfoRequest request) {
+    public ErrorCode modifyUserInfo(String userId, WFCMessage.ModifyMyInfoRequest request) throws Exception {
         HazelcastInstance hzInstance = m_Server.getHazelcastInstance();
         IMap<String, WFCMessage.User> mUserMap = hzInstance.getMap(USERS);
 
@@ -1480,8 +1480,8 @@ public class MemoryMessagesStore implements IMessagesStore {
         if (modified) {
             builder.setUpdateDt(System.currentTimeMillis());
             user = builder.build();
+            databaseStore.updateUser(user);
             mUserMap.set(userId, user);
-//            databaseStore.updateUser(user);
             return ErrorCode.ERROR_CODE_SUCCESS;
         } else {
             return ErrorCode.ERROR_CODE_NOT_MODIFIED;
@@ -1526,9 +1526,10 @@ public class MemoryMessagesStore implements IMessagesStore {
     }
 
     @Override
-    public void addUserInfo(WFCMessage.User user, String password) {
+    public void addUserInfo(WFCMessage.User user, String password) throws Exception {
         HazelcastInstance hzInstance = m_Server.getHazelcastInstance();
         IMap<String, WFCMessage.User> mUserMap = hzInstance.getMap(USERS);
+        databaseStore.updateUser(user);
         mUserMap.put(user.getUid(), user);
         databaseStore.updateUserPassword(user.getUid(), password);
     }
@@ -1539,7 +1540,12 @@ public class MemoryMessagesStore implements IMessagesStore {
         IMap<String, WFCMessage.User> mUserMap = hzInstance.getMap(USERS);
         WFCMessage.User us = mUserMap.get(userId);
         if (us != null) {
-            us = WFCMessage.User.newBuilder().setUid(userId).setUpdateDt(System.currentTimeMillis()).setDisplayName("已删除用户").setName(userId).setDeleted(1).build();
+            us = WFCMessage.User.newBuilder().setUid(userId).setUpdateDt(System.currentTimeMillis()).setDisplayName("").setName(userId).setDeleted(1).build();
+            try {
+                databaseStore.updateUser(us);
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
             mUserMap.put(userId, us);
         }
 

broker/src/main/java/io/moquette/persistence/UserLoader.java

@@ -19,7 +19,7 @@ import java.util.Map;
 public class UserLoader implements MapStore<String, WFCMessage.User> {
     @Override
     public void store(String s, WFCMessage.User user) {
-        getDatabaseStore().updateUser(user);
+//        getDatabaseStore().updateUser(user);
     }
 
     @Override

broker/src/main/java/io/moquette/spi/IMessagesStore.java

@@ -129,13 +129,13 @@ public interface IMessagesStore {
     WFCMessage.Robot getRobot(String robotId);
     void addRobot(WFCMessage.Robot robot);
     ErrorCode getUserInfo(List<WFCMessage.UserRequest> requestList, WFCMessage.PullUserResult.Builder builder);
-    ErrorCode modifyUserInfo(String userId, WFCMessage.ModifyMyInfoRequest request);
+    ErrorCode modifyUserInfo(String userId, WFCMessage.ModifyMyInfoRequest request) throws Exception;
 
     ErrorCode modifyUserStatus(String userId, int status);
     int getUserStatus(String userId);
     List<InputOutputUserBlockStatus> getUserStatusList();
 
-    void addUserInfo(WFCMessage.User user, String password);
+    void addUserInfo(WFCMessage.User user, String password) throws Exception;
     void destoryUser(String userId);
     WFCMessage.User getUserInfo(String userId);
     WFCMessage.User getUserInfoByName(String name);