This article is excerpted from the book “Spring Cloud Microservice Introduction and Progress”.
1 Configure real-time push design after publishing
One of the most important features of configuration center is real-time push, and because of this feature, we can rely on configuration center to do a lot of things. In the Smconf configuration center developed by myself, Smconf relies on the Watch mechanism of Zookeeper to realize real-time push.
The figure above gives a brief overview of the configuration publishing process:
- Users can edit and publish configurations on the Portal
- The Portal invokes the interface provided by the Admin Service to publish data
- After receiving the request, the Admin Service sends ReleaseMessage to each Config Service to notify the Config Service of the configuration change
- After receiving a ReleaseMessage, the Config Service notifies the corresponding client based on Http long connection
2 Implementation of Sending ReleaseMessage
ReleaseMessage messages are a simple message queue implemented by Mysql. The reason for not using message-oriented middleware is to make Apollo deployment as simple as possible with minimal external dependencies.
The figure above briefly describes the process of sending a ReleaseMessage:
- The Admin Service inserts a message record into the ReleaseMessage table after the configuration is published
- The Config Service starts a thread periodically scanning the ReleaseMessage table to see if there are new message records
- The Config Service notifies all message listeners when it finds a new message record
- The message listener notifies the corresponding client when it receives the configuration published information
3 Config Service Notifies the client
Notification is based on Http long connection implementation, mainly divided into the following steps:
- The client initiates an Http request to the Notifications /v2 interface of the Config Service
- The V2 interface suspends the request through Spring DeferredResult and does not return immediately
- If no configuration of interest to the client is published within 60 seconds, Http status code 304 is returned to the client
- If a configuration change is found, the setResult method of DeferredResult is called, passing in namespace information with the configuration change, and the request is returned immediately
- The client requests the Config Service to obtain the latest configuration of the namespace after obtaining the newly configured namespace
4 Source code parsing real-time push design
Apollo push code is a lot of code, so I don’t have to analyze it in detail in this book, but I’ve simplified the push code a little bit to explain it to you, so it’s easier to understand. Of course, I’m going to keep it simple, so I’m not going to worry about a lot of details, just to get you to understand the core principles of Apollo push.
The logic for sending ReleaseMessage is to write a simple interface, store it in a queue, and call this interface when testing to simulate configuration updates and send ReleaseMessage messages.
@restController Public Class NotificationControllerV2 implements ReleaseMessageListener { Public static Queue<String> Queue = new LinkedBlockingDeque<>(); @GetMapping("/addMsg")
public String addMsg() {
queue.add("xxx");
return "success"; }}Copy the codeAfter the message is sent, the Config Service starts a thread to scan the ReleaseMessage table periodically to see if there are any new message records.
@Component public class ReleaseMessageScanner implements InitializingBean { @Autowired private NotificationControllerV2 configController; @override public void afterPropertiesSet() throws Exception {new Thread(() -> {for (;;) {
String result = NotificationControllerV2.queue.poll();
if (result != null) {
ReleaseMessage message = new ReleaseMessage();
message.setMessage(result);
configController.handleMessage(message);
}
}
}).start();;
}
}
Copy the codeThe loop reads the queue in NotificationControllerV2, constructs a ReleaseMessage object if there is a message, and calls the handleMessage() method in NotificationControllerV2 to process the message.
ReleaseMessage is a single field that simulates message content:
public class ReleaseMessage {
private String message;
public void setMessage(String message) {
this.message = message;
}
public String getMessage() {
returnmessage; }}Copy the codeNext, what does handleMessage do
NotificationControllerV2 implements the ReleaseMessageListener interface, in which the handleMessage() method is defined.
public interface ReleaseMessageListener {
void handleMessage(ReleaseMessage message);
}
Copy the codeHandleMessage is a message listener that notifies clients when configuration changes occur. When the message listener receives the information published by the configuration, it notifies the corresponding client:
@RestController public class NotificationControllerV2 implements ReleaseMessageListener { private final Multimap<String, DeferredResultWrapper> deferredResults = Multimaps .synchronizedSetMultimap(HashMultimap.create()); @Override public void handleMessage(ReleaseMessage message) { System.err.println("handleMessage:"+ message);
List<DeferredResultWrapper> results = Lists.newArrayList(deferredResults.get("xxxx"));
for (DeferredResultWrapper deferredResultWrapper : results) {
List<ApolloConfigNotification> list = new ArrayList<>();
list.add(new ApolloConfigNotification("application", 1)); deferredResultWrapper.setResult(list); }}}Copy the codeApollo’s real-time push is implemented based on Spring DeferredResult, and as you can see from the handleMessage() method, we fetch deferredResults through deferredResults. DeferredResults is the first line of the Multimap, the Key is the message content, and the Value is the business wrapper class DeferredResultWrapper for DeferredResult. Let’s look at the code for DeferredResultWrapper:
public class DeferredResultWrapper { private static final long TIMEOUT = 60 * 1000; // 60 seconds private static final ResponseEntity<List<ApolloConfigNotification>> NOT_MODIFIED_RESPONSE_LIST = new ResponseEntity<>(HttpStatus.NOT_MODIFIED); private DeferredResult<ResponseEntity<List<ApolloConfigNotification>>> result; publicDeferredResultWrapper() {
result = new DeferredResult<>(TIMEOUT, NOT_MODIFIED_RESPONSE_LIST);
}
public void onTimeout(Runnable timeoutCallback) {
result.onTimeout(timeoutCallback);
}
public void onCompletion(Runnable completionCallback) {
result.onCompletion(completionCallback);
}
public void setResult(ApolloConfigNotification notification) {
setResult(Lists.newArrayList(notification));
}
public void setResult(List<ApolloConfigNotification> notifications) {
result.setResult(new ResponseEntity<>(notifications, HttpStatus.OK));
}
public DeferredResult<ResponseEntity<List<ApolloConfigNotification>>> getResult() {
returnresult; }}Copy the codeThe setResult() method is set to return the result to the client. This is how the client is notified via message listeners when the configuration changes. When is the client connected?
@restController Public Class NotificationControllerV2 implements ReleaseMessageListener { Public static Queue<String> Queue = new LinkedBlockingDeque<>(); private final Multimap<String, DeferredResultWrapper> deferredResults = Multimaps .synchronizedSetMultimap(HashMultimap.create()); @GetMapping("/getConfig")
public DeferredResult<ResponseEntity<List<ApolloConfigNotification>>> getConfig() {
DeferredResultWrapper deferredResultWrapper = new DeferredResultWrapper();
List<ApolloConfigNotification> newNotifications = getApolloConfigNotifications();
if(! CollectionUtils.isEmpty(newNotifications)) { deferredResultWrapper.setResult(newNotifications); }else {
deferredResultWrapper.onTimeout(() -> {
System.err.println("onTimeout");
});
deferredResultWrapper.onCompletion(() -> {
System.err.println("onCompletion");
});
deferredResults.put("xxxx", deferredResultWrapper);
}
return deferredResultWrapper.getResult();
}
private List<ApolloConfigNotification> getApolloConfigNotifications() {
List<ApolloConfigNotification> list = new ArrayList<>();
String result = queue.poll();
if(result ! = null) { list.add(new ApolloConfigNotification("application", 1));
}
returnlist; }}Copy the codeNotificationControllerV2 provides a/getConfig interface, the client will invoke the interface at start-up time, this time will perform getApolloConfigNotifications () method to get if there is any change in the configuration information, If any prove configuration changes, directly through deferredResultWrapper. SetResult (newNotifications); The result is returned to the client. After receiving the result, the client pulls the configuration information and overwrites the local configuration.
If getApolloConfigNotifications () method returns information configuration changes, prove configuration did not change, will be added to the deferredResults DeferredResultWrapper object, Wait for the message listener to notify when subsequent configuration changes occur.
At the same time, the request is suspended and will not be returned immediately. This is done with the following code in DeferredResultWrapper:
private static final long TIMEOUT = 60 * 1000; // 60 seconds private static final ResponseEntity<List<ApolloConfigNotification>> NOT_MODIFIED_RESPONSE_LIST = new ResponseEntity<>(HttpStatus.NOT_MODIFIED); private DeferredResult<ResponseEntity<List<ApolloConfigNotification>>> result; publicDeferredResultWrapper() {
result = new DeferredResult<>(TIMEOUT, NOT_MODIFIED_RESPONSE_LIST);
}
Copy the codeWhen the DeferredResult object is created, it specifies the timeout period and the response code to be returned. If there is no message listener within 60 seconds, the request will time out and the client will receive the response code 304.
Config Service (Config) {Config Service (Config) {Config Service (Config) {Config Service (Config)}}
public class ClientTest {
public static void main(String[] args) {
reg();
}
private static void reg() {
System.err.println("Registered");
String result = request("http://localhost:8081/getConfig");
if(result ! = null) {// Configuration updated, pull configuration again //...... } // re-register reg(); } private static String request(String url) { HttpURLConnection connection = null; BufferedReader reader = null; try { URL getUrl = new URL(url); connection = (HttpURLConnection) getUrl.openConnection(); connection.setReadTimeout(90000); connection.setConnectTimeout(3000); connection.setRequestMethod("GET");
connection.setRequestProperty("Accept-Charset"."utf-8");
connection.setRequestProperty("Content-Type"."application/json");
connection.setRequestProperty("Charset"."UTF-8");
System.out.println(connection.getResponseCode());
if (200 == connection.getResponseCode()) {
reader = new BufferedReader(new InputStreamReader(connection.getInputStream(), "UTF-8"));
StringBuilder result = new StringBuilder();
String line = null;
while((line = reader.readLine()) ! = null) { result.append(line); } System.out.println("The results" + result);
return result.toString();
}
} catch (IOException e) {
e.printStackTrace();
} finally {
if (connection != null) {
connection.disconnect();
}
}
returnnull; }}Copy the codeFirst start the service where the /getConfig interface resides, and then start the client. The client will initiate a registration request. If there is any change, the client will directly obtain the result and update the configuration. If no modification is made, the request will be suspended. The read timeout period set on the client is 90 seconds, longer than the 60 seconds timeout period set on the server.
Each time you receive a result, you must re-register whether it has been modified or not. In this way, you can achieve the effect of configuring real-time push.
We can simulate a configuration change by calling the /addMsg interface we wrote earlier, and the client will get the result immediately.
This article is excerpted from the book “Spring Cloud Microservice Introduction and Progress” **.
Last year, the book “Spring Cloud Microservices: Full Stack Technology and Case Analysis” ** was published, which received everyone’s support and feedback. Based on everyone’s feedback, we corrected and improved it again.
Based on the relatively stable Spring Cloud finchley. SR2 version and Spring Boot 2.0.6.RELEASE version.
At the same time will be listed code standard archive, the previous are together, not convenient for readers to refer to and run.
It also adds new content like Apollo, Spring Cloud Gateway, and production practices.